Skip to content

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_address(token: str) -> str

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(token: str) -> int

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

get_fresh_swap_transaction(
    metadata: dict[str, Any],
) -> dict[str, Any]

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

EnsoClient(config: EnsoConfig)

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

chain_id property

chain_id: int

Get the chain ID.

wallet_address property

wallet_address: str

Get the configured wallet address.

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_from_params(
    params: RouteParams,
) -> RouteTransaction

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_router_address(
    routing_strategy: RoutingStrategy | None = None,
) -> str

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_chain_id(chain: str | int) -> int

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

chain_id property

chain_id: int

Get numeric chain ID.

__post_init__

__post_init__() -> None

Validate configuration and resolve API key.

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

EnsoConfigError(message: str, parameter: str | None = None)

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

EnsoValidationError(
    message: str,
    field: str | None = None,
    value: Any | None = None,
)

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

from_api_response(data: dict[str, Any]) -> Hop

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

from_api_response classmethod

from_api_response(data: dict[str, Any]) -> Quote

Create Quote from API response.

get_price_impact_percentage

get_price_impact_percentage() -> float | None

Get price impact as a percentage.

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)

is_cross_chain property

is_cross_chain: bool

Check if this is a cross-chain route.

to_api_format

to_api_format() -> dict[str, Any]

Convert parameters to Enso API format.

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

is_cross_chain property

is_cross_chain: bool

Check if this is a cross-chain route.

from_api_response classmethod

from_api_response(data: dict[str, Any]) -> RouteTransaction

Create RouteTransaction from API response.

get_amount_out_wei

get_amount_out_wei(token_address: str | None = None) -> int

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_percentage() -> float | None

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: str, to: str, from_address: str, value: str
)

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

from_api_response(data: dict[str, Any]) -> Transaction

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_approval_receipt(
    receipt: dict[str, Any],
) -> dict[str, Any]

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