Swapping Tokens

Overview

The CoW Protocol Python SDK provides a high-level swap_tokens function that enables you to execute token swaps with built-in MEV protection through batch auctions. This guide covers the complete swap workflow including token approval and order execution.

Prerequisites

Before swapping tokens, ensure you have:

An Ethereum account with a private key
Sufficient balance of the token you want to sell
Token approval for the CoW Protocol Vault Relayer

Token Approval Flow

Before calling swap_tokens, you must approve the CowContractAddress.VAULT_RELAYER to spend your sell token. The swap will fail without proper approval.

Import Required Modules

from web3 import Web3, Account
from web3.types import Wei
from cowdao_cowpy.cow.swap import swap_tokens
from cowdao_cowpy.common.chains import Chain
from cowdao_cowpy.common.constants import CowContractAddress

Approve Token Spending

Grant the Vault Relayer permission to spend your tokens:

from eth_typing import ChecksumAddress

# Initialize Web3 and account
w3 = Web3(Web3.HTTPProvider('YOUR_RPC_URL'))
account = Account.from_key('YOUR_PRIVATE_KEY')

# Token addresses
sell_token: ChecksumAddress = Web3.to_checksum_address(
    "0xbe72E441BF55620febc26715db68d3494213D8Cb"  # USDC
)

# ERC20 ABI for approve function
erc20_abi = [
    {
        "constant": False,
        "inputs": [
            {"name": "spender", "type": "address"},
            {"name": "amount", "type": "uint256"}
        ],
        "name": "approve",
        "outputs": [{"name": "", "type": "bool"}],
        "type": "function"
    }
]

# Create contract instance
token_contract = w3.eth.contract(address=sell_token, abi=erc20_abi)

# Approve Vault Relayer
vault_relayer = CowContractAddress.VAULT_RELAYER.value
approve_amount = Wei(2**256 - 1)  # Max approval

tx = token_contract.functions.approve(
    vault_relayer,
    approve_amount
).build_transaction({
    'from': account.address,
    'nonce': w3.eth.get_transaction_count(account.address),
})

signed_tx = account.sign_transaction(tx)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

Execute the Swap

Once approval is complete, you can execute the swap.

Basic Token Swap

Here’s a complete example of swapping tokens:

import os
import asyncio
from dotenv import load_dotenv
from web3 import Account, Web3
from web3.types import Wei
from cowdao_cowpy.cow.swap import swap_tokens
from cowdao_cowpy.common.chains import Chain

# Token addresses (Sepolia testnet)
BUY_TOKEN = Web3.to_checksum_address(
    "0xfFf9976782d46CC05630D1f6eBAb18b2324d6B14"  # WETH
)
SELL_TOKEN = Web3.to_checksum_address(
    "0xbe72E441BF55620febc26715db68d3494213D8Cb"  # USDC
)

# Amount to sell (50 USDC with 18 decimals)
SELL_AMOUNT = Wei(50_000_000_000_000_000_000)

load_dotenv()
PRIVATE_KEY = os.getenv("PRIVATE_KEY")
ACCOUNT = Account.from_key(PRIVATE_KEY)

async def main():
    completed_order = await swap_tokens(
        amount=SELL_AMOUNT,
        account=ACCOUNT,
        chain=Chain.SEPOLIA,
        sell_token=SELL_TOKEN,
        buy_token=BUY_TOKEN,
    )

    print(f"Order UID: {completed_order.uid}")
    print(f"Order URL: {completed_order.url}")

if __name__ == "__main__":
    asyncio.run(main())

Advanced Configuration

The swap_tokens function accepts several optional parameters for advanced use cases:

# Custom slippage tolerance (1%)
completed_order = await swap_tokens(
    amount=SELL_AMOUNT,
    account=ACCOUNT,
    chain=Chain.MAINNET,
    sell_token=SELL_TOKEN,
    buy_token=BUY_TOKEN,
    slippage_tolerance=0.01,
)

# Custom order validity
import time

valid_until = int(time.time()) + 3600  # Valid for 1 hour

completed_order = await swap_tokens(
    amount=SELL_AMOUNT,
    account=ACCOUNT,
    chain=Chain.MAINNET,
    sell_token=SELL_TOKEN,
    buy_token=BUY_TOKEN,
    valid_to=valid_until,
)

# Partially fillable order
completed_order = await swap_tokens(
    amount=SELL_AMOUNT,
    account=ACCOUNT,
    chain=Chain.MAINNET,
    sell_token=SELL_TOKEN,
    buy_token=BUY_TOKEN,
    partially_fillable=True,
)

# Safe multi-sig order
from eth_typing import ChecksumAddress

safe_address: ChecksumAddress = Web3.to_checksum_address("0x...")

completed_order = await swap_tokens(
    amount=SELL_AMOUNT,
    account=ACCOUNT,
    chain=Chain.MAINNET,
    sell_token=SELL_TOKEN,
    buy_token=BUY_TOKEN,
    safe_address=safe_address,
)

# Custom app data
from cowdao_cowpy.app_data.utils import generate_app_data

app_data_result = generate_app_data(
    app_code="my-trading-app",
    graffiti="My Custom Message"
)

completed_order = await swap_tokens(
    amount=SELL_AMOUNT,
    account=ACCOUNT,
    chain=Chain.MAINNET,
    sell_token=SELL_TOKEN,
    buy_token=BUY_TOKEN,
    app_data=app_data_result.app_data_hash.root,
)

Function Parameters

Parameter	Type	Required	Default	Description
`amount`	`Wei`	Yes	-	Amount of sell token to swap
`account`	`LocalAccount`	Yes	-	Ethereum account for signing
`chain`	`Chain`	Yes	-	Target blockchain network
`sell_token`	`ChecksumAddress`	Yes	-	Address of token to sell
`buy_token`	`ChecksumAddress`	Yes	-	Address of token to buy
`safe_address`	`ChecksumAddress \| None`	No	`None`	Safe/multisig address (if applicable)
`app_data`	`str`	No	Default hash	Custom app data hash
`valid_to`	`int \| None`	No	Quote validity	Order expiration (Unix timestamp)
`env`	`Envs`	No	`"prod"`	Environment (`"prod"`, `"staging"`)
`slippage_tolerance`	`float`	No	`0.005`	Maximum slippage (0.005 = 0.5%)
`partially_fillable`	`bool`	No	`False`	Allow partial order fills

Return Value

The function returns a CompletedOrder object with:

@dataclass
class CompletedOrder:
    uid: UID  # Unique order identifier
    url: str  # CoW Explorer URL for tracking

How It Works

Quote Request

The function requests a quote from the OrderBook API with your sell amount and tokens.

Order Construction

Creates an Order object with sell/buy amounts (with slippage protection), validity period, token balances configuration, and app data metadata.

Order Signing

Signs the order using EIP-712 typed data. For regular accounts: ECDSA signature. For Safe addresses: PreSign signature.

Order Submission

Posts the signed order to the OrderBook API for solvers to execute.

After the Swap

When swap_tokens returns a CompletedOrder, the order has been submitted to the CoW Protocol orderbook. Solvers now compete in batch auctions to find the best execution path for your trade.

Monitoring Order Status

You can poll the order status using the OrderBookApi:

from cowdao_cowpy.order_book.api import OrderBookApi
from cowdao_cowpy.order_book.config import OrderBookAPIConfigFactory
from cowdao_cowpy.common.config import SupportedChainId

api = OrderBookApi(
    OrderBookAPIConfigFactory.get_config("prod", SupportedChainId.MAINNET)
)

order = await api.get_order_by_uid(completed_order.uid)
print(f"Status: {order.status}")  # open, fulfilled, cancelled, expired

Order Status Values

Status	Description
`open`	Order is active and waiting for solvers to fill it
`fulfilled`	Order has been fully executed
`cancelled`	Order was cancelled by the user
`expired`	Order reached its `valid_to` timestamp without being filled
`presignaturePending`	Safe/multisig order awaiting on-chain pre-signature

Cancelling an Order

If your order has not yet been filled, you can request cancellation:

await api.delete_order(completed_order.uid, signature, signing_scheme)

For a complete guide on querying, tracking, and cancelling orders, see Managing Orders.

Supported Chains

The SDK supports swapping on the following networks:

Ethereum Mainnet — Chain.MAINNET
Gnosis Chain — Chain.GNOSIS
Arbitrum One — Chain.ARBITRUM_ONE
Base — Chain.BASE
Sepolia Testnet — Chain.SEPOLIA

Use Chain.SEPOLIA for testing without spending real funds. Get testnet tokens from Sepolia faucets.

Error Handling

import asyncio
from cowdao_cowpy.common.api.errors import UnexpectedResponseError

async def safe_swap():
    try:
        completed_order = await swap_tokens(
            amount=SELL_AMOUNT,
            account=ACCOUNT,
            chain=Chain.MAINNET,
            sell_token=SELL_TOKEN,
            buy_token=BUY_TOKEN,
        )
        print(f"Swap successful: {completed_order.url}")
    except UnexpectedResponseError as e:
        print(f"API error: {e}")
    except ValueError as e:
        print(f"Invalid parameter: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

asyncio.run(safe_swap())

Common Issues

Problem	Likely Cause	Fix
`InsufficientAllowance` error	Vault Relayer not approved	Run the token approval flow first
`InsufficientBalance` error	Not enough sell token	Check balance before calling `swap_tokens`
Order created but never fills	Low liquidity or tight slippage	Increase `slippage_tolerance`, try common pairs
`NoLiquidity` error	Token pair not supported	Verify token addresses, try larger amounts
`UnexpectedResponseError`	API issue or invalid params	Check token addresses are checksummed, amount > 0
Order expires	Solvers couldn’t fill in time	Use longer `valid_to`, wider slippage

Next Steps

Learn about Managing Orders to track and cancel swaps
Explore App Data to add custom metadata
Check Contract Interaction for advanced signing scenarios

Getting Started

Concepts

Guides

Advanced

API Reference

Overview

Prerequisites

Token Approval Flow

Basic Token Swap

Advanced Configuration

Function Parameters

Return Value

How It Works

After the Swap

Monitoring Order Status

Order Status Values

Cancelling an Order

Supported Chains

Error Handling

Common Issues

Next Steps

​Overview

​Prerequisites

​Token Approval Flow

​Basic Token Swap

​Advanced Configuration

​Function Parameters

​Return Value

​How It Works

​After the Swap

​Monitoring Order Status

​Order Status Values

​Cancelling an Order

​Supported Chains

​Error Handling

​Common Issues

​Next Steps

Overview

Prerequisites

Token Approval Flow

Basic Token Swap

Advanced Configuration

Function Parameters

Return Value

How It Works

After the Swap

Monitoring Order Status

Order Status Values

Cancelling an Order

Supported Chains

Error Handling

Common Issues

Next Steps