Biconomy integration

Biconomy.md

@@ -0,0 +1,23 @@
+# Biconomy Smart Accounts
+
+## Description
+AutoTx can be used with a Biconomy Smart Account.
+
+## Getting started
+First, add the private key of the account you want to use as the owner to the `.env` file. Use the variable `SMART_ACCOUNT_OWNER_PK`.
+Since the private key is stored as plain text, it is recommended to use a test account.
+
+Next, start the `smart_account_api`, run: `poetry run start-smart-account-api`. This API server will be used to interact with the Biconomy Smart Account.
+You can run `poetry run stop-smart-account-api` to stop the server.
+
+To start AutoTx, you can now run: `poetry run ask "my-prompt-here"`.
+
+When running for the first time, you will be prompted to send funds to your new Biconomy Smart Account in order to pay for the gas fees.
+```
+Using Biconomy smart account: {SMART-ACCOUNT-ADDRESS}
+Detected empty account balance.
+To use your new smart account, please top it up with some native currency.
+Send the funds to: {SMART-ACCOUNT-ADDRESS} on {NETWORK-NAME}
+Waiting...
+```
+After sending the funds, AutoTx will automatically detect the balance and continue with the transaction.

README.md

@@ -158,6 +158,8 @@ python benchmarks.py ./autotx/tests/file_name.py::function_name 5
 # run a specific test with 5 iterations and name the output folder (instead of the default timestamp)
 python benchmarks.py ./autotx/tests/file_name.py::function_name 5 output_folder_name
 ```
+# Biconomy Smart Accounts
+To view the Biconomy Smart Accounts documentation, please see the [Biconomy.md](./Biconomy.md) file.
 
 # API Server
 To view the API server documentation, please see the [API.md](./API.md) file.

autotx/AutoTx.py

@@ -19,7 +19,7 @@
 from autotx.utils.logging.Logger import Logger
 from autotx.utils.ethereum.networks import NetworkInfo
 from autotx.utils.constants import OPENAI_BASE_URL, OPENAI_MODEL_NAME
-from autotx.wallets.smart_wallet import SmartWallet
+from autotx.smart_accounts.smart_account import SmartAccount
 
 @dataclass(kw_only=True)
 class CustomModel:
@@ -64,7 +64,7 @@ class RunResult:
 
 class AutoTx:
     web3: Web3
-    wallet: SmartWallet
+    wallet: SmartAccount
     logger: Logger
     intents: list[Intent]
     network: NetworkInfo
@@ -82,7 +82,7 @@ class AutoTx:
     def __init__(
         self,
         web3: Web3,
-        wallet: SmartWallet,
+        wallet: SmartAccount,
         network: NetworkInfo,
         agents: list[AutoTxAgent],
         config: Config,

autotx/cli.py

@@ -1,17 +1,30 @@
 from dotenv import load_dotenv
+
+from autotx.eth_address import ETHAddress
 load_dotenv()
 
+from eth_account import Account
+import time
+from web3 import Web3
 import uuid
 import uvicorn
 from typing import cast
 import click
+import uuid
+from eth_account.signers.local import LocalAccount
 
-from autotx.wallets.safe_smart_wallet import SafeSmartWallet
+from autotx.utils.ethereum.get_native_balance import get_native_balance
+from autotx.utils.ethereum.networks import NetworkInfo
+from autotx.utils.constants import SMART_ACCOUNT_OWNER_PK
+from autotx.smart_accounts.safe_smart_account import SafeSmartAccount
+from autotx.smart_accounts.smart_account import SmartAccount
 from autotx.utils.configuration import AppConfig
 from autotx.utils.is_dev_env import is_dev_env
 from autotx.setup import print_agent_address, setup_agents
 from autotx.AutoTx import AutoTx, Config
 from autotx.utils.ethereum.helpers.show_address_balances import show_address_balances
+from autotx.smart_accounts.smart_account import SmartAccount
+from autotx.smart_accounts.local_biconomy_smart_account import LocalBiconomySmartAccount
 
 def print_autotx_info() -> None:
     print("""
@@ -32,6 +45,15 @@ def print_autotx_info() -> None:
 def main() -> None:
     pass
 
+def wait_for_native_top_up(web3: Web3, address: ETHAddress) -> None:
+    network = NetworkInfo(web3.eth.chain_id)
+
+    print(f"Detected empty account balance.\nTo use your new smart account, please top it up with some native currency.\nSend the funds to: {address} on {network.chain_id.name}")
+    print("Waiting...")
+    while get_native_balance(web3, address) == 0:
+        time.sleep(2)
+    print(f"Account balance detected ({get_native_balance(web3, address)}). Ready to use.")
+
 @main.command()
 @click.argument('prompt', required=False)
 @click.option("-n", "--non-interactive", is_flag=True, help="Non-interactive mode (will not expect further user input or approval)")
@@ -45,8 +67,17 @@ def run(prompt: str | None, non_interactive: bool, verbose: bool, logs: str | No
     if prompt == None:
         prompt = click.prompt("What do you want to do?")
 
-    app_config = AppConfig.load(fill_dev_account=True)
-    wallet = SafeSmartWallet(app_config.manager, auto_submit_tx=non_interactive)
+    app_config = AppConfig()
+    wallet: SmartAccount
+    if SMART_ACCOUNT_OWNER_PK:
+        smart_account_owner = cast(LocalAccount, Account.from_key(SMART_ACCOUNT_OWNER_PK))
+        wallet = LocalBiconomySmartAccount(app_config.web3, smart_account_owner, auto_submit_tx=non_interactive)
+        print(f"Using Biconomy smart account: {wallet.address}")
+        if get_native_balance(app_config.web3, wallet.address) == 0:
+            wait_for_native_top_up(app_config.web3, wallet.address)
+    else:
+        wallet = SafeSmartAccount(app_config.rpc_url, app_config.network_info, auto_submit_tx=non_interactive, fill_dev_account=True)
+        print("Using Safe smart account: {wallet.address}")
 
     (get_llm_config, agents, logs_dir) = setup_agents(logs, cache)
 

autotx/server.py

@@ -12,12 +12,13 @@
 from autotx import db
 from autotx.AutoTx import AutoTx, Config as AutoTxConfig
 from autotx.intents import Intent
+from autotx.smart_accounts.smart_account import SmartAccount
 from autotx.transactions import Transaction
 from autotx.utils.configuration import AppConfig
 from autotx.utils.ethereum.chain_short_names import CHAIN_ID_TO_SHORT_NAME
 from autotx.utils.ethereum.networks import SUPPORTED_NETWORKS_CONFIGURATION_MAP
-from autotx.wallets.api_smart_wallet import ApiSmartWallet
-from autotx.wallets.smart_wallet import SmartWallet
+from autotx.smart_accounts.api_smart_account import ApiSmartAccount
+from autotx.smart_accounts.safe_smart_account import SafeSmartAccount
 
 class AutoTxParams:
     verbose: bool
@@ -61,17 +62,17 @@ def authorize(authorization: str | None) -> models.App:
 
     return app
 
-def load_config_for_user(app_id: str, user_id: str, address: str, chain_id: int) -> AppConfig:
+def load_wallet_for_user(app_config: AppConfig, app_id: str, user_id: str, address: str) -> SmartAccount:
     agent_private_key = db.get_agent_private_key(app_id, user_id)
 
     if not agent_private_key:
         raise HTTPException(status_code=400, detail="User not found")
 
     agent = Account.from_key(agent_private_key)
 
-    app_config = AppConfig.load(smart_account_addr=address, subsidized_chain_id=chain_id, agent=agent)
+    wallet = SafeSmartAccount(app_config.rpc_url, app_config.network_info, auto_submit_tx=False, smart_account_addr=address, agent=agent)
 
-    return app_config
+    return wallet
 
 def authorize_app_and_user(authorization: str | None, user_id: str) -> tuple[models.App, models.AppUser]:
     app = authorize(authorization)
@@ -86,15 +87,16 @@ async def build_transactions(app_id: str, user_id: str, chain_id: int, address:
     if task.running:
         raise HTTPException(status_code=400, detail="Task is still running")
 
-    app_config = load_config_for_user(app_id, user_id, address, chain_id)
+    app_config = AppConfig(subsidized_chain_id=chain_id)
+    wallet = load_wallet_for_user(app_config, app_id, user_id, address)
 
     if task.intents is None or len(task.intents) == 0:
         return []
 
     transactions: list[Transaction] = []
 
     for intent in task.intents:
-        transactions.extend(await intent.build_transactions(app_config.web3, app_config.network_info, app_config.manager.address))
+        transactions.extend(await intent.build_transactions(app_config.web3, app_config.network_info, wallet.address))
 
     return transactions
 
@@ -113,17 +115,14 @@ async def create_task(task: models.TaskCreate, background_tasks: BackgroundTasks
 
     prompt = task.prompt
 
-    app_config = AppConfig.load(smart_account_addr=task.address, subsidized_chain_id=task.chain_id)
+    app_config = AppConfig(subsidized_chain_id=task.chain_id)
 
-    wallet: SmartWallet
-    if autotx_params.is_dev:
-        wallet = ApiSmartWallet(app_config.web3, app_config.manager, tasks)
-    else:
-        wallet = ApiSmartWallet(app_config.web3, app_config.manager, tasks)
+    wallet = SafeSmartAccount(app_config.rpc_url, app_config.network_info, smart_account_addr=task.address)
+    api_wallet = ApiSmartAccount(app_config.web3, wallet, tasks)
 
-    created_task: models.Task = tasks.start(prompt, wallet.address.hex, app_config.network_info.chain_id.value, app_user.id)
+    created_task: models.Task = tasks.start(prompt, api_wallet.address.hex, app_config.network_info.chain_id.value, app_user.id)
     task_id = created_task.id
-    wallet.task_id = task_id
+    api_wallet.task_id = task_id
 
     try:
         (get_llm_config, agents, logs_dir) = setup.setup_agents(autotx_params.logs, cache=autotx_params.cache)
@@ -138,7 +137,7 @@ def on_notify_user(message: str) -> None:
 
         autotx = AutoTx(
             app_config.web3,
-            wallet,
+            api_wallet,
             app_config.network_info,
             agents,
             AutoTxConfig(verbose=autotx_params.verbose, get_llm_config=get_llm_config, logs_dir=logs_dir, max_rounds=autotx_params.max_rounds),
@@ -296,12 +295,10 @@ def send_transactions(
         return f"https://app.safe.global/transactions/queue?safe={CHAIN_ID_TO_SHORT_NAME[str(chain_id)]}:{address}"
 
     try:
-        app_config = load_config_for_user(app.id, user_id, address, chain_id)
+        app_config = AppConfig(subsidized_chain_id=chain_id)
+        wallet = load_wallet_for_user(app_config, app.id, user_id, address)
 
-        app_config.manager.send_multisend_tx_batch(
-            transactions,
-            require_approval=False,
-        )
+        wallet.send_transactions(transactions)
     except SafeAPIException as e:
         if "is not an owner or delegate" in str(e):
             raise HTTPException(status_code=400, detail="Agent is not an owner or delegate")
@@ -339,7 +336,9 @@ async def get_version() -> Dict[str, str]:
 
 def setup_server(verbose: bool, logs: str | None, max_rounds: int | None, cache: bool, is_dev: bool, check_valid_safe: bool) -> None:
     if is_dev: 
-        AppConfig.load(check_valid_safe=check_valid_safe) # Loading the configuration deploys the dev wallet in dev mode
+        app_config = AppConfig() 
+        # Loading the SafeSmartAccount will deploy a new Safe if one is not already deployed
+        SafeSmartAccount(app_config.rpc_url, app_config.network_info, fill_dev_account=True, check_valid_safe=check_valid_safe)
 
     global autotx_params
     autotx_params = AutoTxParams(

autotx/setup.py

@@ -10,9 +10,7 @@
 from autotx.autotx_agent import AutoTxAgent
 from autotx.utils.constants import COINGECKO_API_KEY, OPENAI_API_KEY, OPENAI_BASE_URL, OPENAI_MODEL_NAME
 from autotx.utils.ethereum import SafeManager
-from autotx.utils.ethereum.cached_safe_address import get_cached_safe_address
 from autotx.eth_address import ETHAddress
-from autotx.utils.ethereum.helpers.fill_dev_account_with_tokens import fill_dev_account_with_tokens
 from autotx.utils.ethereum.helpers.get_dev_account import get_dev_account
 from autotx.utils.ethereum.helpers.show_address_balances import show_address_balances
 from autotx.utils.ethereum.networks import NetworkInfo
@@ -23,7 +21,7 @@ def print_agent_address() -> None:
     acc = get_or_create_agent_account()
     print(f"Agent address: {acc.address}")
 
-def setup_safe(smart_account_addr: ETHAddress | None, agent: LocalAccount, client: EthereumClient, fill_dev_account: bool, check_valid_safe: bool) -> SafeManager:
+def setup_safe(smart_account_addr: ETHAddress | None, agent: LocalAccount, client: EthereumClient, check_valid_safe: bool) -> SafeManager:
     web3 = client.w3
 
     chain_id = web3.eth.chain_id
@@ -51,13 +49,8 @@ def setup_safe(smart_account_addr: ETHAddress | None, agent: LocalAccount, clien
         print("No smart account connected, deploying a new one...")
         dev_account = get_dev_account()
 
-        is_safe_deployed = get_cached_safe_address()
         manager = SafeManager.deploy_safe(client, dev_account, agent, [dev_account.address, agent.address], 1)
         print(f"Smart account deployed: {manager.address}")
-
-        if not is_safe_deployed and fill_dev_account:
-            fill_dev_account_with_tokens(client, dev_account, manager.address, network_info)
-            print(f"Funds sent to smart account for testing purposes")
 
         print("=" * 50)
         print("Starting smart account balances:")

autotx/smart_account_api.py

@@ -0,0 +1,33 @@
+import subprocess
+import sys
+
+container_name = "smart_account_api"
+
+def start() -> None:
+    build = subprocess.run(
+        ["docker", "build", "-t", "smart_account_api", ".", "-f", "smart-account-api.Dockerfile"], capture_output=True
+    )
+
+    if build.returncode != 0:
+        sys.exit(
+            "Local node start up has failed. Make sure you have docker desktop installed and running"
+        )
+
+    subprocess.run(
+        [
+            "docker",
+            "run",
+            "-d",
+            "--name",
+            container_name,
+            "-p",
+            "7080:7080",
+            "--env-file",
+            ".env",
+            "smart_account_api",
+        ],
+        check=True,
+    )
+
+def stop() -> None:
+    subprocess.run(["docker", "container", "rm", container_name, "-f"], check=True)

autotx/smart_accounts/api_smart_account.py

@@ -0,0 +1,37 @@
+from web3 import Web3
+from autotx import db
+from autotx.intents import Intent
+from autotx.smart_accounts.smart_account import SmartAccount
+from autotx.transactions import TransactionBase
+
+# ApiSmartAccount is a wrapper around other SmartAccounts that allows for hooks to be added to the transaction process
+class ApiSmartAccount(SmartAccount):
+    wallet: SmartAccount
+    tasks: db.TasksRepository
+    task_id: str | None
+
+    def __init__(self, web3: Web3, wallet: SmartAccount, tasks: db.TasksRepository, task_id: str | None = None):
+        super().__init__(web3, wallet.address)
+        self.wallet = wallet
+        self.tasks = tasks
+        self.task_id = task_id
+
+    def on_intents_prepared(self, intents: list[Intent]) -> None:
+        if self.task_id is None:
+            raise ValueError("Task ID is required")
+
+        saved_task = self.tasks.get(self.task_id)
+        if saved_task is None:
+            raise ValueError("Task not found")
+
+        saved_task.intents.extend(intents)
+        self.tasks.update(saved_task)
+
+    async def on_intents_ready(self, _intents: list[Intent]) -> bool | str:
+        return True
+
+    def send_transaction(self, transaction: TransactionBase) -> None:
+        self.wallet.send_transaction(transaction)
+
+    def send_transactions(self, transactions: list[TransactionBase]) -> None:
+        self.wallet.send_transactions(transactions)