DEV Community

Naren karthi
Naren karthi

Posted on

How to Build an Autonomous Trading Agent with Python

#coding #tutorial #web3 #ai

🤖🔗 Tutorial: Build an Autonomous AI Agent with a Crypto Wallet

Project: "CryptoBuddy" – A CLI-based AI Agent that can read blockchain state, propose transactions (swaps, transfers, contract interactions), and execute them on Ethereum Sepolia Testnet after user confirmation.

Difficulty: Intermediate

Time: ~60–90 Minutes

Cost: $0 (Testnet only)

🎯 1. Project Overview & Architecture

What we are building

A Python application where you chat with an LLM (GPT-4o / Llama 3). The LLM has Tools (functions) to:

  1. Read: Check ERC20 balances, ETH balance, Token prices (via Coingecko), Contract state.
  2. Write: Build unsigned transactions (Transfer, Approve, Swap on Uniswap V2).
  3. Execute: Present the transaction to the user → User signs with private key (locally) → Broadcast to Sepolia.

Architecture Diagram 

[ User Chat Input ]
       │
       ▼
[ LangChain Agent (LLM + Prompt + Memory) ]
       │
       ├─► [ Read Tools ] ──► [ RPC Node (Alchemy/Infura) ] ──► Blockchain State
       │
       └─► [ Write Tools ] ──► [ Build Tx Object (Dict) ] ──► [ User Confirms ] ──► [ Sign & Broadcast ]
Enter fullscreen mode Exit fullscreen mode

Tech Stack

Layer Technology
Language Python 3.10+
AI Framework langchain, langchain-openai, langchain-community
Blockchain web3.py, eth-account, eth-typing
Network Ethereum Sepolia Testnet
RPC Provider Alchemy / Infura / QuickNode (Free Tier)
DEX Uniswap V2 (Standard Router on Sepolia)

✅ 2. Prerequisites

2.1 Accounts & Keys (Get these before coding)

  1. GitHub Account (for cloning).
  2. OpenAI API Key (or Groq/Together.ai for free/fast Llama 3).
  3. Alchemy/Infura Account → Create App → Copy HTTPS Sepolia RPC URL.
  4. Sepolia ETH → Get from Sepolia Faucet or Alchemy Faucet.
  5. Test Tokens:
    • USDC (Sepolia): 0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238 (Circle official).
    • WETH (Sepolia): 0x7b79995e5f793A07Bc00c21412e50Ecae098E7f9.
    • Get test USDC from Circle Faucet or swap Sepolia ETH → WETH → USDC on Uniswap Sepolia UI.

2.2 Local Environment 

# 1. Install Python 3.10+ (Check: python3 --version)
# 2. Create Project Folder
mkdir crypto-ai-agent && cd crypto-ai-agent

# 3. Create Virtual Environment (Highly Recommended)
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 4. Install Dependencies
pip install langchain langchain-openai langchain-community web3 python-dotenv eth-account requests tiktoken
Enter fullscreen mode Exit fullscreen mode

📁 3. Project Structure 

crypto-ai-agent/
├── .env                  # SECRETS (NEVER COMMIT THIS)
├── config.py             # Constants: Addresses, ABIs, RPC URLs
├── tools/
│   ├── __init__.py
│   ├── read_tools.py     # Balance checks, Price feeds, Contract reads
│   └── write_tools.py    # Build Tx: Transfer, Approve, Swap
├── agent/
│   ├── __init__.py
│   └── core.py           # Agent Initialization, Prompt, Memory
├── utils/
│   ├── __init__.py
│   └── tx_helper.py      # Signing, Broadcasting, Gas Estimation
└── main.py               # Entry Point: CLI Chat Loop
Enter fullscreen mode Exit fullscreen mode

⚙️ 4. Configuration & Constants (config.py)

Store ABIs and Addresses here to keep main code clean.

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

# --- Network & Keys ---
RPC_URL = os.getenv("SEPOLIA_RPC_URL")
PRIVATE_KEY = os.getenv("PRIVATE_KEY") # User's wallet PK (for signing)
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# --- Addresses (Sepolia) ---
WETH_ADDRESS = "0x7b79995e5f793A07Bc00c21412e50Ecae098E7f9"
USDC_ADDRESS = "0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"
UNISWAP_V2_ROUTER_ADDRESS = "0xC532a74256D3Db42D0Bf7a0400fEFDbad7694008" # Sepolia Uniswap V2 Router

# --- Minimal ABIs ---
ERC20_ABI = [
    {"inputs":[{"name":"account","type":"address"}],"name":"balanceOf","outputs":[{"name":"","type":"uint256"}],"stateMutability":"view","type":"function"},
    {"inputs":[{"name":"spender","type":"address"},{"name":"amount","type":"uint256"}],"name":"approve","outputs":[{"name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},
    {"inputs":[{"name":"spender","type":"address"}],"name":"allowance","outputs":[{"name":"","type":"uint256"}],"stateMutability":"view","type":"function"},
    {"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"stateMutability":"view","type":"function"},
    {"inputs":[],"name":"symbol","outputs":[{"name":"","type":"string"}],"stateMutability":"view","type":"function"}
]

UNISWAP_V2_ROUTER_ABI = [
    # swapExactTokensForTokens
    {"inputs":[{"name":"amountIn","type":"uint256"},{"name":"amountOutMin","type":"uint256"},{"name":"path","type":"address[]"},{"name":"to","type":"address"},{"name":"deadline","type":"uint256"}],"name":"swapExactTokensForTokens","outputs":[{"name":"amounts","type":"uint256[]"}],"stateMutability":"nonpayable","type":"function"},
    # swapExactETHForTokens (Payable)
    {"inputs":[{"name":"amountOutMin","type":"uint256"},{"name":"path","type":"address[]"},{"name":"to","type":"address"},{"name":"deadline","type":"uint256"}],"name":"swapExactETHForTokens","outputs":[{"name":"amounts","type":"uint256[]"}],"stateMutability":"payable","type":"function"},
    # getAmountsOut (View)
    {"inputs":[{"name":"amountIn","type":"uint256"},{"name":"path","type":"address[]"}],"name":"getAmountsOut","outputs":[{"name":"amounts","type":"uint256[]"}],"stateMutability":"view","type":"function"}
]

# --- Safety Limits ---
MAX_TX_VALUE_ETH = 0.1  # Agent refuses to build tx > 0.1 ETH value without explicit override
GAS_BUFFER = 1.2        # Multiply estimated gas by 1.2
Enter fullscreen mode Exit fullscreen mode

🛠️ 5. Utility Layer: Transaction Helper (utils/tx_helper.py)

Handles the "Plumbing": Nonce management, Gas estimation, Signing, Broadcasting.

# utils/tx_helper.py
from web3 import Web3
from web3.types import TxParams, Wei
from eth_account import Account
from eth_account.signers.local import LocalAccount
from config import RPC_URL, PRIVATE_KEY, GAS_BUFFER
import math

w3 = Web3(Web3.HTTPProvider(RPC_URL))
account: LocalAccount = Account.from_key(PRIVATE_KEY)
USER_ADDRESS = account.address

def get_nonce() -> int:
    return w3.eth.get_transaction_count(USER_ADDRESS, 'pending')

def estimate_gas(tx_params: TxParams) -> int:
    try:
        estimated = w3.eth.estimate_gas(tx_params)
        return math.ceil(estimated * GAS_BUFFER)
    except Exception as e:
        print(f"⚠️ Gas Estimation Failed: {e}. Using default 300,000")
        return 300_000

def build_base_tx(value_eth: float = 0) -> TxParams:
    """Returns a base transaction dict with nonce, gasPrice, chainId, from."""
    return {
        'from': USER_ADDRESS,
        'nonce': get_nonce(),
        'chainId': 11155111, # Sepolia Chain ID
        'value': w3.to_wei(value_eth, 'ether'),
        'maxFeePerGas': w3.to_wei('5', 'gwei'), # Sepolia base fee usually low
        'maxPriorityFeePerGas': w3.to_wei('1.5', 'gwei'),
    }

def sign_and_broadcast(tx_params: TxParams) -> str:
    """Signs tx with private key and sends. Returns Tx Hash."""
    # 1. Estimate Gas
    gas_limit = estimate_gas(tx_params)
    tx_params['gas'] = gas_limit

    print(f"\n📝 **Transaction Summary**")
    print(f"   To: {tx_params.get('to')}")
    print(f"   Value: {w3.from_wei(tx_params.get('value', 0), 'ether')} ETH")
    print(f"   Gas Limit: {gas_limit}")
    print(f"   Data: {tx_params.get('data', '0x')[:50]}...")

    confirm = input("\n⚠️  Sign and Broadcast this transaction? (y/N): ").strip().lower()
    if confirm != 'y':
        return "User Cancelled"

    # 2. Sign
    signed_tx = account.sign_transaction(tx_params)

    # 3. Broadcast
    tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)
    tx_hash_hex = tx_hash.hex()
    print(f"🚀 Broadcasted! Hash: {tx_hash_hex}")
    print(f"🔗 View on Etherscan: https://sepolia.etherscan.io/tx/{tx_hash_hex}")

    # 4. Wait for Receipt (Optional but good UX)
    print("⏳ Waiting for confirmation...")
    receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
    status = "✅ Success" if receipt.status == 1 else "❌ Failed"
    print(f"{status} | Block: {receipt.blockNumber} | Gas Used: {receipt.gasUsed}")

    return tx_hash_hex
Enter fullscreen mode Exit fullscreen mode

📖 6. Read Tools (tools/read_tools.py)

These are View functions. They don't cost gas. The LLM calls these to "see" the world.


python
# tools/read_tools.py
from langchain.tools import tool
from web3 import Web3
from config import RPC_URL, ERC20_ABI, UNISWAP_V2_ROUTER_ABI, WETH_ADDRESS, USDC_ADDRESS, UNISWAP_V2_ROUTER_ADDRESS
import requests
import json

w3 = Web3(Web3.HTTPProvider(RPC_URL))

# Helper to load contract
def get_contract(address, abi):
    return w3.eth.contract(address=Web3.to_checksum_address(address), abi=abi)

@tool
def get_eth_balance(address: str) -> str:
    """Get native ETH balance of an address. Input: Ethereum Address (string)."""
    try:
        addr = Web3.to_checksum_address(address)
        balance_wei = w3.eth.get_balance(addr)
        return f"{w3.from_wei(balance_wei, 'ether'):.4f} ETH"
    except Exception as e:
        return f"Error: {e}"

@tool
def get_erc20_balance(token_address: str, wallet_address: str) -> str:
    """Get ERC20 token balance. Inputs: Token Contract Address, Wallet Address."""
    try:
        contract = get_contract(token_address, ERC20_ABI)
        bal = contract.functions.balanceOf(Web3.to_checksum_address(wallet_address)).call()
        decimals = contract.functions.decimals().call()
        symbol = contract.functions.symbol().call()
        return f"{bal / (10**decimals):.4f} {symbol}"
    except Exception as e:
        return f"Error: {e}"

@tool
def get_token_price_usd(token_symbol: str) -> str:
    """Get current USD price from CoinGecko. Input: Token Symbol (e.g., 'eth', 'usdc', 'weth')."""
    try:
        # Map symbols to CoinGecko IDs
        id_map = {"eth": "ethereum", "weth": "weth", "usdc": "usd-coin"}
        cg_id = id_map.get(token_symbol.lower(), token_symbol.lower())
        url = f"https://api.coingecko.com/api/v3/simple/price?ids={cg_id}&vs_currencies=usd"
        resp = requests.get(url, timeout=10).json()
        price = resp.get(cg_id, {}).get('usd', 'N/A')
        return f"${price:,.2f}" if isinstance(price, (int, float)) else "Price not found"
    except Exception as e:
        return f"Error fetching price: {e}"

@tool
def check_allowance(token_address: str, owner_address: str, spender_address: str) -> str:
    """Check how much token 'owner' allowed 'spender' to use. Inputs: Token, Owner, Spender."""
    try:
        contract = get_contract(token_address, ERC20_ABI)
        allowance = contract.functions.allowance(
            Web3.to_checksum_address(owner_address), 
            Web3.to_checksum_address(spender_address)
        ).call()
        decimals = contract.functions.decimals().call()
        return f"{allowance / (10**decimals):.4f}"
    except Exception as e:
        return f"Error: {e}"

@tool
def simulate_swap_amount_out(token_in: str, token_out: str, amount_in_human: float) -> str:
    """Simulate Uniswap V2 Swap Output. Inputs: TokenIn Address, TokenOut Address, AmountIn (Human readable)."""
    try:
        router = get_contract(UNISWAP_V2_ROUTER_ADDRESS, UNISWAP_V2_ROUTER_ABI)
        token_in_contract = get_contract(token_in,

#coding #tutorial #web3 #AI
Enter fullscreen mode Exit fullscreen mode

Top comments (0)