Enso¶
Connector for Enso Finance DEX aggregator.
almanak.framework.connectors.enso
¶
Enso Finance Protocol Connector.
Enso is a routing and composable transaction protocol that aggregates liquidity across multiple DEXs and lending protocols.
This connector provides: - EnsoClient: SDK for interacting with the Enso Finance API - EnsoAdapter: Adapter for converting intents to Enso transactions - EnsoReceiptParser: Parser for extracting results from transaction receipts
Supports both same-chain and cross-chain swaps via Enso's bridge aggregation (Stargate, LayerZero).
Example
from almanak.framework.connectors.enso import EnsoClient, EnsoAdapter, EnsoConfig
Create client¶
config = EnsoConfig( api_key="your-api-key", chain="base", wallet_address="0x...", ) client = EnsoClient(config)
Same-chain swap¶
route = client.get_route( token_in="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", # USDC on Base token_out="0x4200000000000000000000000000000000000006", # WETH on Base amount_in=1000000000, # 1000 USDC slippage_bps=50, # 0.5% )
Cross-chain swap: Base -> Arbitrum¶
cross_chain_route = client.get_cross_chain_route( token_in="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", # USDC on Base token_out="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH on Arbitrum amount_in=1000000000, destination_chain="arbitrum", )
EnsoAdapter
¶
EnsoAdapter(
config: EnsoConfig,
use_safe_route_single: bool = True,
price_provider: dict[str, Decimal] | None = None,
allow_placeholder_prices: bool = False,
token_resolver: TokenResolver | None = None,
)
Adapter for Enso protocol integration with the Intent system.
This adapter converts high-level SwapIntents into executable transactions using the Enso Finance routing protocol.
Features: - Multi-DEX routing for optimal swap prices - Automatic slippage protection via safeRouteSingle - ERC-20 approval handling
Example
config = EnsoConfig(chain="arbitrum", wallet_address="0x...") adapter = EnsoAdapter(config)
intent = SwapIntent( from_token="USDC", to_token="WETH", amount_usd=Decimal("1000"), ) bundle = adapter.compile_swap_intent(intent)
Initialize the Enso adapter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
EnsoConfig
|
Enso client configuration |
required |
use_safe_route_single
|
bool
|
Whether to transform routes to use safeRouteSingle for slippage protection (default True) |
True
|
price_provider
|
dict[str, Decimal] | None
|
Price oracle dict (token symbol -> USD price). Required for production use to calculate accurate slippage amounts. |
None
|
allow_placeholder_prices
|
bool
|
If False (default), raises ValueError when no price_provider is given. Set to True ONLY for unit tests. |
False
|
token_resolver
|
TokenResolver | None
|
Optional TokenResolver instance for unified token resolution. If None, uses the default singleton from get_token_resolver(). |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If no price_provider is provided and allow_placeholder_prices is False. |
resolve_token_address
¶
Resolve token symbol or address to address using TokenResolver.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
Token symbol (e.g., "USDC") or address |
required |
Returns:
| Type | Description |
|---|---|
str
|
Token address |
Raises:
| Type | Description |
|---|---|
TokenResolutionError
|
If the token cannot be resolved |
get_token_decimals
¶
Get token decimals using TokenResolver.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
Token symbol or address |
required |
Returns:
| Type | Description |
|---|---|
int
|
Token decimals |
Raises:
| Type | Description |
|---|---|
TokenResolutionError
|
If decimals cannot be determined |
compile_swap_intent
¶
compile_swap_intent(
intent: SwapIntent,
price_oracle: dict[str, Decimal] | None = None,
) -> ActionBundle
Compile a SwapIntent to an ActionBundle using Enso.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
intent
|
SwapIntent
|
The SwapIntent to compile |
required |
price_oracle
|
dict[str, Decimal] | None
|
Optional price oracle for USD conversions |
None
|
Returns:
| Type | Description |
|---|---|
ActionBundle
|
ActionBundle containing transactions for execution |
get_fresh_swap_transaction
¶
Fetch fresh swap transaction data immediately before execution.
IMPORTANT: Enso route data becomes stale within seconds. This method should be called immediately before executing a swap transaction to get fresh route data from the Enso API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
metadata
|
dict[str, Any]
|
The metadata from a compiled ActionBundle containing route_params |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Fresh transaction data dict with keys: - to: Router address - value: Native token value (usually 0) - data: Fresh route calldata - gas_estimate: Raw gas estimate (orchestrator applies buffer) - amount_out: Expected output amount (for logging/verification) |
Raises:
| Type | Description |
|---|---|
ValueError
|
If metadata doesn't contain route_params |
Exception
|
If route fetching fails |
Example
After executing approval, fetch fresh swap data¶
fresh_tx = adapter.get_fresh_swap_transaction(bundle.metadata)
Build and sign transaction with fresh data¶
unsigned_tx = UnsignedTransaction( to=fresh_tx["to"], data=fresh_tx["data"], value=fresh_tx["value"], gas_limit=fresh_tx["gas_estimate"], ... )
EnsoClient
¶
Client for interacting with the Enso Finance API.
This client provides methods for: - Getting swap routes across multiple DEXs - Getting quotes for swaps - Building bundle transactions for complex DeFi operations - Approving tokens for the Enso router
Example
config = EnsoConfig( chain="arbitrum", wallet_address="0x...", ) client = EnsoClient(config)
Get a simple swap route¶
route = client.get_route( token_in="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", token_out="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", amount_in=1000000000, slippage_bps=50, )
Initialize the Enso client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
EnsoConfig
|
Enso client configuration |
required |
get_route
¶
get_route(
token_in: str,
token_out: str,
amount_in: int,
slippage_bps: int = 50,
from_address: str | None = None,
receiver: str | None = None,
routing_strategy: RoutingStrategy | None = None,
max_price_impact_bps: int | None = None,
destination_chain_id: int | None = None,
refund_receiver: str | None = None,
) -> RouteTransaction
Get the best swap route from one token to another.
Supports both same-chain swaps and cross-chain swaps via Enso's bridge aggregation (Stargate, LayerZero).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
Input token address |
required |
token_out
|
str
|
Output token address |
required |
amount_in
|
int
|
Input amount in wei (as integer) |
required |
slippage_bps
|
int
|
Slippage tolerance in basis points (default 50 = 0.5%) |
50
|
from_address
|
str | None
|
Address executing the swap (defaults to config wallet) |
None
|
receiver
|
str | None
|
Address to receive output (defaults to from_address) |
None
|
routing_strategy
|
RoutingStrategy | None
|
Routing strategy to use |
None
|
max_price_impact_bps
|
int | None
|
Maximum allowed price impact in basis points |
None
|
destination_chain_id
|
int | None
|
Target chain ID for cross-chain swaps (None for same-chain) |
None
|
refund_receiver
|
str | None
|
Address to receive refunds if cross-chain fails (defaults to from_address) |
None
|
Returns:
| Type | Description |
|---|---|
RouteTransaction
|
RouteTransaction with transaction data and route details |
Raises:
| Type | Description |
|---|---|
EnsoAPIError
|
If the API request fails |
PriceImpactExceedsThresholdError
|
If price impact exceeds threshold |
Example
Same-chain swap on Arbitrum¶
route = client.get_route( token_in="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", # USDC token_out="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH amount_in=1000000000, )
Cross-chain swap: Base -> Arbitrum¶
route = client.get_route( token_in="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", # USDC on Base token_out="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH on Arbitrum amount_in=1000000000, destination_chain_id=42161, # Arbitrum )
get_route_from_params
¶
Get route using RouteParams object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
params
|
RouteParams
|
Route parameters |
required |
Returns:
| Type | Description |
|---|---|
RouteTransaction
|
RouteTransaction with transaction data |
get_quote
¶
get_quote(
token_in: str,
token_out: str,
amount_in: int,
from_address: str | None = None,
routing_strategy: RoutingStrategy | None = None,
destination_chain_id: int | None = None,
) -> Quote
Get a quote for a swap without building the transaction.
Supports both same-chain and cross-chain quotes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
Input token address |
required |
token_out
|
str
|
Output token address |
required |
amount_in
|
int
|
Input amount in wei |
required |
from_address
|
str | None
|
Address executing the swap |
None
|
routing_strategy
|
RoutingStrategy | None
|
Routing strategy to use |
None
|
destination_chain_id
|
int | None
|
Target chain ID for cross-chain quotes (None for same-chain) |
None
|
Returns:
| Type | Description |
|---|---|
Quote
|
Quote with expected output amount |
get_bundle
¶
get_bundle(
bundle_actions: list[BundleAction],
from_address: str | None = None,
routing_strategy: RoutingStrategy | None = None,
skip_quote: bool = False,
) -> dict[str, Any]
Get a bundle transaction for multiple DeFi actions.
Bundles allow composing multiple operations (deposits, borrows, swaps) into a single transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bundle_actions
|
list[BundleAction]
|
List of actions to bundle |
required |
from_address
|
str | None
|
Address executing the bundle |
None
|
routing_strategy
|
RoutingStrategy | None
|
Routing strategy to use |
None
|
skip_quote
|
bool
|
Skip quote generation (for operations that don't need it) |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Bundle response with transaction data |
get_approval
¶
get_approval(
token_address: str,
amount: int | None = None,
from_address: str | None = None,
routing_strategy: RoutingStrategy | None = None,
) -> dict[str, Any]
Get approval transaction data for a token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_address
|
str
|
Token to approve |
required |
amount
|
int | None
|
Amount to approve (defaults to unlimited) |
None
|
from_address
|
str | None
|
Address granting approval |
None
|
routing_strategy
|
RoutingStrategy | None
|
Routing strategy (determines spender) |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Approval transaction data |
get_router_address
¶
Get the Enso router address for the current chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
routing_strategy
|
RoutingStrategy | None
|
Routing strategy (router or delegate) |
None
|
Returns:
| Type | Description |
|---|---|
str
|
Router contract address |
resolve_chain_id
staticmethod
¶
Resolve a chain name or ID to a chain ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
chain
|
str | int
|
Chain name (e.g., "arbitrum") or chain ID (e.g., 42161) |
required |
Returns:
| Type | Description |
|---|---|
int
|
Chain ID as integer |
Raises:
| Type | Description |
|---|---|
EnsoValidationError
|
If chain name is not supported |
get_cross_chain_route
¶
get_cross_chain_route(
token_in: str,
token_out: str,
amount_in: int,
destination_chain: str | int,
slippage_bps: int = 50,
receiver: str | None = None,
max_price_impact_bps: int | None = None,
) -> RouteTransaction
Convenience method for cross-chain swaps.
This is a simplified interface for cross-chain operations that handles chain ID resolution and sets sensible defaults.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
Input token address (on source chain) |
required |
token_out
|
str
|
Output token address (on destination chain) |
required |
amount_in
|
int
|
Input amount in wei |
required |
destination_chain
|
str | int
|
Destination chain name (e.g., "arbitrum") or chain ID |
required |
slippage_bps
|
int
|
Slippage tolerance in basis points (default 50 = 0.5%) |
50
|
receiver
|
str | None
|
Address to receive output (defaults to wallet address) |
None
|
max_price_impact_bps
|
int | None
|
Maximum allowed price impact in basis points |
None
|
Returns:
| Type | Description |
|---|---|
RouteTransaction
|
RouteTransaction with cross-chain transaction data |
Example
Bridge USDC from Base to Arbitrum (swap to WETH on arrival)¶
client = EnsoClient(EnsoConfig(chain="base", wallet_address="0x...")) route = client.get_cross_chain_route( token_in="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", # USDC on Base token_out="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH on Arbitrum amount_in=1000 * 10**6, # 1000 USDC destination_chain="arbitrum", )
EnsoConfig
dataclass
¶
EnsoConfig(
chain: str,
wallet_address: str,
api_key: str | None = None,
base_url: str = "https://api.enso.finance",
routing_strategy: RoutingStrategy = RoutingStrategy.ROUTER,
timeout: int = 30,
)
Configuration for Enso client.
Attributes:
| Name | Type | Description |
|---|---|---|
api_key |
str | None
|
Enso API key (or set ENSO_API_KEY env var) |
chain |
str
|
Chain name (e.g., "arbitrum", "ethereum") or chain ID |
wallet_address |
str
|
Default wallet address for transactions |
base_url |
str
|
Enso API base URL |
routing_strategy |
RoutingStrategy
|
Default routing strategy |
timeout |
int
|
Request timeout in seconds |
EnsoAPIError
¶
EnsoAPIError(
message: str,
status_code: int,
endpoint: str | None = None,
error_data: dict | None = None,
)
Bases: EnsoError
Exception raised for errors in the API response.
Attributes:
| Name | Type | Description |
|---|---|---|
message |
Error message |
|
status_code |
HTTP status code of the response |
|
endpoint |
The API endpoint that was called |
|
error_type |
Classified error type (e.g., SERVER_ERROR, RATE_LIMIT) |
|
api_error_message |
The specific error message from the API response |
|
error_data |
Parsed error data from the response, if available |
EnsoConfigError
¶
Bases: EnsoError
Exception raised for SDK configuration errors.
Attributes:
| Name | Type | Description |
|---|---|---|
message |
Error message |
|
parameter |
Name of the configuration parameter that caused the error |
EnsoError
¶
Bases: Exception
Base exception class for all Enso SDK errors.
EnsoValidationError
¶
Bases: EnsoError
Exception raised for validation errors.
Attributes:
| Name | Type | Description |
|---|---|---|
message |
Error message |
|
field |
Name of the field that failed validation |
|
value |
The invalid value |
PriceImpactExceedsThresholdError
¶
PriceImpactExceedsThresholdError(
message: str,
price_impact_bps: float | None = None,
threshold_bps: int | None = None,
)
Bases: EnsoError
Raised when route price impact exceeds maximum threshold.
Attributes:
| Name | Type | Description |
|---|---|---|
message |
Error message |
|
price_impact_bps |
Actual price impact in basis points |
|
threshold_bps |
Maximum allowed price impact in basis points |
Hop
dataclass
¶
Hop(
token_in: list[str] = list(),
token_out: list[str] = list(),
protocol: str = "",
action: str = "",
primary: str = "",
)
A single hop in a swap route.
Attributes:
| Name | Type | Description |
|---|---|---|
token_in |
list[str]
|
Input token addresses |
token_out |
list[str]
|
Output token addresses |
protocol |
str
|
Protocol used for this hop |
action |
str
|
Action performed |
primary |
str
|
Primary address (pool/contract) |
from_api_response
classmethod
¶
Create Hop from API response.
Quote
dataclass
¶
Quote(
amount_out: str,
gas: str | None = None,
price_impact: float | None = None,
route: list[Hop] | None = None,
chain_id: int | None = None,
)
Quote response from Enso API.
Attributes:
| Name | Type | Description |
|---|---|---|
amount_out |
str
|
Expected output amount |
gas |
str | None
|
Estimated gas |
price_impact |
float | None
|
Price impact in basis points |
route |
list[Hop] | None
|
List of hops in the route |
chain_id |
int | None
|
Chain ID for this quote |
RouteParams
dataclass
¶
RouteParams(
from_address: str,
token_in: str,
token_out: str,
amount_in: int,
chain_id: int,
slippage_bps: int = 50,
routing_strategy: RoutingStrategy = RoutingStrategy.ROUTER,
receiver: str | None = None,
max_price_impact_bps: int | None = None,
destination_chain_id: int | None = None,
refund_receiver: str | None = None,
)
Parameters for getting a swap route from Enso.
Attributes:
| Name | Type | Description |
|---|---|---|
from_address |
str
|
Address executing the transaction |
token_in |
str
|
Input token address |
token_out |
str
|
Output token address |
amount_in |
int
|
Input amount in wei (as int) |
chain_id |
int
|
Source blockchain chain ID |
slippage_bps |
int
|
Slippage tolerance in basis points (e.g., 50 = 0.5%) |
routing_strategy |
RoutingStrategy
|
Routing strategy to use |
receiver |
str | None
|
Address to receive output tokens (defaults to from_address) |
max_price_impact_bps |
int | None
|
Maximum allowed price impact in basis points |
destination_chain_id |
int | None
|
Target chain ID for cross-chain swaps (None for same-chain) |
refund_receiver |
str | None
|
Address to receive refunds if cross-chain fails (required for cross-chain) |
RouteTransaction
dataclass
¶
RouteTransaction(
gas: str,
tx: Transaction,
amount_out: dict[str, Any],
price_impact: float | None = None,
route: list[Hop] = list(),
fee_amount: list[str] = list(),
created_at: int | None = None,
chain_id: int | None = None,
destination_chain_id: int | None = None,
bridge_fee: str | None = None,
estimated_time: int | None = None,
)
Route transaction response from Enso API.
Attributes:
| Name | Type | Description |
|---|---|---|
gas |
str
|
Estimated gas for the transaction |
tx |
Transaction
|
Transaction data |
amount_out |
dict[str, Any]
|
Expected output amount (as dict with token -> amount) |
price_impact |
float | None
|
Price impact in basis points |
route |
list[Hop]
|
List of hops in the route |
fee_amount |
list[str]
|
Fee amounts |
created_at |
int | None
|
Timestamp when route was created |
chain_id |
int | None
|
Source chain ID for this route |
destination_chain_id |
int | None
|
Destination chain ID for cross-chain routes (None for same-chain) |
bridge_fee |
str | None
|
Bridge fee for cross-chain routes (in native token wei) |
estimated_time |
int | None
|
Estimated completion time in seconds for cross-chain routes |
from_api_response
classmethod
¶
Create RouteTransaction from API response.
get_amount_out_wei
¶
Get the output amount in wei.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_address
|
str | None
|
Specific token address to get amount for. If None, returns the first/only amount. |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Output amount in wei as integer |
get_price_impact_percentage
¶
Get price impact as a percentage.
Returns:
| Type | Description |
|---|---|
float | None
|
Price impact as a percentage (e.g., 3.0 for 3%) or None if not available. |
RoutingStrategy
¶
Bases: str, Enum
Enso routing strategies.
Transaction
dataclass
¶
Transaction data from Enso API response.
Attributes:
| Name | Type | Description |
|---|---|---|
data |
str
|
Encoded calldata |
to |
str
|
Target contract address |
from_address |
str
|
Sender address |
value |
str
|
Native token value to send (in wei) |
from_api_response
classmethod
¶
Create Transaction from API response.
EnsoReceiptParser
¶
Parser for Enso transaction receipts.
This parser extracts swap results from transaction receipts by: 1. Checking transaction status 2. Parsing Transfer event logs to find amounts 3. Identifying the recipient's received amount
Example
parser = EnsoReceiptParser() receipt = web3.eth.get_transaction_receipt(tx_hash)
result = parser.parse_swap_receipt( receipt=receipt, wallet_address="0x...", token_out="0x...", ) print(f"Received: {result.amount_out}")
parse_swap_receipt
¶
parse_swap_receipt(
receipt: dict[str, Any],
wallet_address: str,
token_out: str,
token_in: str | None = None,
expected_amount_out: int | None = None,
) -> SwapResult
Parse a swap transaction receipt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
receipt
|
dict[str, Any]
|
Transaction receipt from web3 |
required |
wallet_address
|
str
|
Address that received the output tokens |
required |
token_out
|
str
|
Output token address |
required |
token_in
|
str | None
|
Input token address (optional) |
None
|
expected_amount_out
|
int | None
|
Expected output amount for validation |
None
|
Returns:
| Type | Description |
|---|---|
SwapResult
|
SwapResult with parsed data |
parse_approval_receipt
¶
Parse an approval transaction receipt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
receipt
|
dict[str, Any]
|
Transaction receipt from web3 |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with approval result |