Aller au contenu

LiFi

Connector for LiFi cross-chain bridge aggregation protocol.

almanak.framework.connectors.lifi

LiFi Cross-Chain Aggregator Connector.

LiFi is a cross-chain liquidity meta-aggregator that routes through bridges (Across, Stargate, Hop, Wormhole, etc.) and DEXs (1inch, 0x, Paraswap, etc.) to find optimal routes for cross-chain and same-chain swaps.

This connector provides: - LiFiClient: SDK for interacting with the LiFi API - LiFiAdapter: Adapter for converting intents to LiFi transactions - LiFiReceiptParser: Parser for extracting results from transaction receipts

Supports both cross-chain bridges and same-chain swaps. Uses standard ERC-20 approvals (no Permit2 needed).

Example

from almanak.framework.connectors.lifi import LiFiClient, LiFiAdapter, LiFiConfig

config = LiFiConfig(chain_id=42161, wallet_address="0x...") client = LiFiClient(config)

Cross-chain swap: Arbitrum USDC -> Base USDC

quote = client.get_quote( from_chain_id=42161, to_chain_id=8453, from_token="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", to_token="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", from_amount="1000000000", from_address="0x...", )

Same-chain swap: Arbitrum USDC -> WETH

same_chain_quote = client.get_quote( from_chain_id=42161, to_chain_id=42161, from_token="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", to_token="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", from_amount="1000000000", from_address="0x...", )

LiFiAdapter

LiFiAdapter(
    config: LiFiConfig,
    price_provider: dict[str, Decimal] | None = None,
    allow_placeholder_prices: bool = False,
    token_resolver: TokenResolver | None = None,
)

Adapter for LiFi protocol integration with the Intent system.

This adapter converts high-level SwapIntents into executable transactions using the LiFi cross-chain aggregation protocol. Supports both same-chain swaps and cross-chain bridge+swap operations.

Features: - Cross-chain swap aggregation (bridge + DEX routing) - Same-chain DEX aggregation - Standard ERC-20 approval (no Permit2) - Deferred transaction pattern (fresh routes at execution)

Example

config = LiFiConfig(chain_id=42161, wallet_address="0x...") adapter = LiFiAdapter(config)

intent = SwapIntent( from_token="USDC", to_token="WETH", amount=Decimal("1000"), ) bundle = adapter.compile_swap_intent(intent)

Initialize the LiFi adapter.

参数:

名称 类型 描述 默认
config LiFiConfig

LiFi client configuration

必需
price_provider dict[str, Decimal] | None

Price oracle dict (token symbol -> USD price). Required for production use to calculate accurate amounts from USD.

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

引发:

类型 描述
ValueError

If no price_provider is provided and allow_placeholder_prices is False.

resolve_token_address

resolve_token_address(
    token: str, chain: str | None = None
) -> str

Resolve token symbol or address to address using TokenResolver.

参数:

名称 类型 描述 默认
token str

Token symbol (e.g., "USDC") or address

必需
chain str | None

Chain name override (defaults to config chain)

None

返回:

类型 描述
str

Token address

引发:

类型 描述
TokenResolutionError

If the token cannot be resolved

get_token_decimals

get_token_decimals(
    token: str, chain: str | None = None
) -> int

Get token decimals using TokenResolver.

参数:

名称 类型 描述 默认
token str

Token symbol or address

必需
chain str | None

Chain name override (defaults to config chain)

None

返回:

类型 描述
int

Token decimals

引发:

类型 描述
TokenResolutionError

If decimals cannot be determined

compile_swap_intent

compile_swap_intent(
    intent: SwapIntent,
    price_oracle: dict[str, Decimal] | None = None,
    destination_chain_id: int | None = None,
) -> ActionBundle

Compile a SwapIntent to an ActionBundle using LiFi.

Supports both same-chain swaps and cross-chain swaps.

参数:

名称 类型 描述 默认
intent SwapIntent

The SwapIntent to compile

必需
price_oracle dict[str, Decimal] | None

Optional price oracle for USD conversions

None
destination_chain_id int | None

Destination chain for cross-chain swaps. If None, uses same chain as source.

None

返回:

类型 描述
ActionBundle

ActionBundle containing transactions for execution

get_fresh_transaction

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

Fetch fresh transaction data immediately before execution.

IMPORTANT: LiFi route data becomes stale quickly. This method should be called immediately before executing a transaction to get fresh route data from the LiFi API.

参数:

名称 类型 描述 默认
metadata dict[str, Any]

The metadata from a compiled ActionBundle containing route_params

必需

返回:

类型 描述
dict[str, Any]

Fresh transaction data dict with keys: - to: Target contract address - value: Native token value - data: Fresh route calldata - gas_estimate: Estimated gas - amount_out: Expected output amount

引发:

类型 描述
ValueError

If metadata doesn't contain route_params

LiFiClient

LiFiClient(config: LiFiConfig)

Client for interacting with the LiFi API.

This client provides methods for: - Getting cross-chain and same-chain swap quotes - Checking cross-chain transfer status - Querying supported tokens and chains

Example

config = LiFiConfig(chain_id=42161, wallet_address="0x...") client = LiFiClient(config)

quote = client.get_quote( from_chain_id=42161, to_chain_id=8453, from_token="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", to_token="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", from_amount="1000000000", from_address="0x...", )

Initialize the LiFi client.

参数:

名称 类型 描述 默认
config LiFiConfig

LiFi client configuration

必需

get_quote

get_quote(
    from_chain_id: int,
    to_chain_id: int,
    from_token: str,
    to_token: str,
    from_amount: str,
    from_address: str,
    to_address: str | None = None,
    slippage: float = 0.005,
    order: LiFiOrderStrategy | None = None,
    allow_bridges: list[str] | None = None,
    deny_bridges: list[str] | None = None,
) -> LiFiStep

Get a quote with executable transaction data.

This calls the LiFi /quote endpoint which returns a single best route with transaction data ready to execute.

参数:

名称 类型 描述 默认
from_chain_id int

Source chain ID

必需
to_chain_id int

Destination chain ID

必需
from_token str

Source token address

必需
to_token str

Destination token address

必需
from_amount str

Amount in smallest unit (wei for ETH, etc.)

必需
from_address str

Sender wallet address

必需
to_address str | None

Receiver address (defaults to from_address)

None
slippage float

Slippage tolerance as decimal (0.005 = 0.5%)

0.005
order LiFiOrderStrategy | None

Route ordering strategy (FASTEST, CHEAPEST, etc.)

None
allow_bridges list[str] | None

Only use these bridges

None
deny_bridges list[str] | None

Exclude these bridges

None

返回:

类型 描述
LiFiStep

LiFiStep with quote details and transaction data

引发:

类型 描述
LiFiAPIError

If the API request fails

LiFiRouteNotFoundError

If no route is found

get_status

get_status(
    tx_hash: str,
    from_chain: int,
    to_chain: int,
    bridge: str | None = None,
) -> LiFiStatusResponse

Check the status of a cross-chain transfer.

参数:

名称 类型 描述 默认
tx_hash str

Source chain transaction hash

必需
from_chain int

Source chain ID

必需
to_chain int

Destination chain ID

必需
bridge str | None

Bridge name (optional, improves lookup speed)

None

返回:

类型 描述
LiFiStatusResponse

LiFiStatusResponse with transfer status

get_tokens

get_tokens(chain_id: int | None = None) -> dict[str, Any]

Get supported tokens.

参数:

名称 类型 描述 默认
chain_id int | None

Filter by chain ID (optional)

None

返回:

类型 描述
dict[str, Any]

Dict of tokens by chain

get_chains

get_chains() -> list[dict[str, Any]]

Get supported chains.

返回:

类型 描述
list[dict[str, Any]]

List of chain information dicts

resolve_chain_id staticmethod

resolve_chain_id(chain: str | int) -> int

Resolve a chain name or ID to a chain ID.

参数:

名称 类型 描述 默认
chain str | int

Chain name (e.g., "arbitrum") or chain ID (e.g., 42161)

必需

返回:

类型 描述
int

Chain ID as integer

引发:

类型 描述
LiFiConfigError

If chain name is not recognized

LiFiConfig dataclass

LiFiConfig(
    chain_id: int,
    wallet_address: str,
    api_key: str | None = None,
    base_url: str = DEFAULT_BASE_URL,
    integrator: str = DEFAULT_INTEGRATOR,
    timeout: int = 30,
    order: LiFiOrderStrategy = LiFiOrderStrategy.RECOMMENDED,
)

Configuration for LiFi client.

属性:

名称 类型 描述
chain_id int

Default source chain ID

wallet_address str

Default wallet address for transactions

api_key str | None

LiFi API key (optional, from LIFI_API_KEY env var)

base_url str

LiFi API base URL

integrator str

Integrator name for LiFi API

timeout int

Request timeout in seconds

order LiFiOrderStrategy

Default route ordering strategy

__post_init__

__post_init__() -> None

Resolve API key from env if not provided.

LiFiAPIError

LiFiAPIError(
    message: str,
    status_code: int,
    endpoint: str | None = None,
    error_data: dict | None = None,
)

Bases: LiFiError

Exception raised for errors in the LiFi API response.

属性:

名称 类型 描述
message

Error message

status_code

HTTP status code of the response

endpoint

The API endpoint that was called

error_type

Classified error type

error_data

Parsed error data from the response

LiFiConfigError

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

Bases: LiFiError

Exception raised for configuration errors.

属性:

名称 类型 描述
message

Error message

parameter

Name of the configuration parameter that caused the error

LiFiError

Bases: Exception

Base exception class for all LiFi connector errors.

LiFiRouteNotFoundError

Bases: LiFiError

Raised when LiFi cannot find a route for the requested transfer.

LiFiTransferFailedError

LiFiTransferFailedError(
    message: str,
    tx_hash: str | None = None,
    status: str | None = None,
    substatus: str | None = None,
)

Bases: LiFiError

Raised when a LiFi cross-chain transfer fails or is refunded.

属性:

名称 类型 描述
message

Error message

tx_hash

Source chain transaction hash

status

LiFi status value (FAILED, etc.)

substatus

LiFi substatus (REFUNDED, PARTIAL, etc.)

LiFiValidationError

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

Bases: LiFiError

Exception raised for validation errors.

属性:

名称 类型 描述
message

Error message

field

Name of the field that failed validation

value

The invalid value

LiFiAction dataclass

LiFiAction(
    from_chain_id: int,
    to_chain_id: int,
    from_token: LiFiToken,
    to_token: LiFiToken,
    from_amount: str,
    from_address: str = "",
    to_address: str = "",
    slippage: float = 0.005,
)

Action details within a LiFi step.

属性:

名称 类型 描述
from_chain_id int

Source chain ID

to_chain_id int

Destination chain ID

from_token LiFiToken

Source token info

to_token LiFiToken

Destination token info

from_amount str

Amount being sent (in wei/smallest unit)

from_address str

Sender address

to_address str

Receiver address

slippage float

Slippage tolerance (0-1)

from_api_response classmethod

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

Create LiFiAction from API response.

LiFiEstimate dataclass

LiFiEstimate(
    from_amount: str = "0",
    to_amount: str = "0",
    to_amount_min: str = "0",
    approval_address: str = "",
    execution_duration: int = 0,
    fee_costs: list[LiFiFeeCost] = list(),
    gas_costs: list[LiFiGasCost] = list(),
)

Estimate information for a LiFi step.

属性:

名称 类型 描述
from_amount str

Input amount (in smallest unit)

to_amount str

Expected output amount

to_amount_min str

Guaranteed minimum output (with slippage)

approval_address str

Contract address to approve tokens to

execution_duration int

Estimated execution time in seconds

fee_costs list[LiFiFeeCost]

List of fee costs

gas_costs list[LiFiGasCost]

List of gas costs

total_gas_estimate property

total_gas_estimate: int

Get total gas estimate across all gas costs.

total_fee_usd property

total_fee_usd: float

Get total fees in USD.

from_api_response classmethod

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

Create from API response.

LiFiFeeCost dataclass

LiFiFeeCost(
    name: str = "",
    description: str = "",
    percentage: str = "0",
    amount: str = "0",
    amount_usd: str = "0",
    token: LiFiToken | None = None,
    included: bool = True,
)

Fee cost information.

属性:

名称 类型 描述
name str

Fee name (e.g., "Bridge Fee")

description str

Fee description

percentage str

Fee as percentage

amount str

Fee amount in token units

amount_usd str

Fee in USD

token LiFiToken | None

Token for fee

included bool

Whether fee is included in the amount

from_api_response classmethod

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

Create from API response.

LiFiGasCost dataclass

LiFiGasCost(
    type: str = "",
    estimate: str = "0",
    limit: str = "0",
    amount: str = "0",
    amount_usd: str = "0",
    token: LiFiToken | None = None,
)

Gas cost information.

属性:

名称 类型 描述
type str

Cost type (e.g., "SUM")

estimate str

Estimated gas units

limit str

Gas limit

amount str

Cost amount in native token (wei)

amount_usd str

Cost in USD

token LiFiToken | None

Native token info

from_api_response classmethod

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

Create from API response.

LiFiOrderStrategy

Bases: StrEnum

LiFi route ordering strategies.

LiFiStatusResponse dataclass

LiFiStatusResponse(
    transaction_id: str = "",
    sending_tx_hash: str = "",
    receiving_tx_hash: str = "",
    bridge_name: str = "",
    from_chain_id: int = 0,
    to_chain_id: int = 0,
    status: str = "",
    substatus: str = "",
    substatus_message: str = "",
)

Status response for a cross-chain transfer.

属性:

名称 类型 描述
transaction_id str

LiFi transaction ID

sending_tx_hash str

Source chain transaction hash

receiving_tx_hash str

Destination chain transaction hash (when complete)

bridge_name str

Bridge used for the transfer

from_chain_id int

Source chain ID

to_chain_id int

Destination chain ID

status str

Overall status (NOT_FOUND, PENDING, DONE, FAILED)

substatus str

Detailed substatus (COMPLETED, PARTIAL, REFUNDED)

substatus_message str

Human-readable substatus message

is_complete property

is_complete: bool

Check if the transfer is complete.

is_failed property

is_failed: bool

Check if the transfer failed.

is_pending property

is_pending: bool

Check if the transfer is still pending.

from_api_response classmethod

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

Create from API response.

LiFiStep dataclass

LiFiStep(
    id: str = "",
    type: str = "",
    tool: str = "",
    action: LiFiAction | None = None,
    estimate: LiFiEstimate | None = None,
    transaction_request: LiFiTransactionRequest
    | None = None,
    included_steps: list[LiFiStep] = list(),
    integrator: str = "",
)

A step in a LiFi route (the main quote response).

This is the top-level response from the /v1/quote endpoint.

属性:

名称 类型 描述
id str

Unique step identifier

type str

Step type (swap, cross, lifi, protocol)

tool str

Bridge/DEX tool used (e.g., "across", "1inch")

action LiFiAction | None

Action details (chains, tokens, amounts)

estimate LiFiEstimate | None

Estimate details (amounts, fees, duration)

transaction_request LiFiTransactionRequest | None

Transaction to execute on-chain

included_steps list[LiFiStep]

Sub-steps for multi-hop routes

integrator str

Integrator identifier

is_cross_chain property

is_cross_chain: bool

Check if this is a cross-chain step.

from_api_response classmethod

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

Create LiFiStep from API response.

get_to_amount

get_to_amount() -> int

Get expected output amount as integer.

get_to_amount_min

get_to_amount_min() -> int

Get guaranteed minimum output amount as integer.

LiFiStepType

Bases: StrEnum

LiFi step types.

LiFiToken dataclass

LiFiToken(
    address: str,
    chain_id: int,
    symbol: str,
    decimals: int,
    name: str = "",
    price_usd: float | None = None,
)

Token information from LiFi API.

属性:

名称 类型 描述
address str

Token contract address

chain_id int

Chain ID where token resides

symbol str

Token symbol (e.g., "USDC")

decimals int

Token decimals

name str

Token name

price_usd float | None

Token price in USD (if available)

from_api_response classmethod

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

Create LiFiToken from API response.

Note: decimals defaults to 18 here for display purposes only. The adapter uses TokenResolver for all amount calculations, so this default does NOT affect financial math.

LiFiTransactionRequest dataclass

LiFiTransactionRequest(
    from_address: str = "",
    to: str = "",
    chain_id: int = 0,
    data: str = "",
    value: str = "0",
    gas_price: str = "0",
    gas_limit: str = "0",
)

Transaction request data from LiFi API.

This contains the actual calldata to execute on-chain.

属性:

名称 类型 描述
from_address str

Sender address

to str

Target contract address (LiFi Diamond)

chain_id int

Chain ID for the transaction

data str

Encoded calldata

value str

Native token value to send (in wei)

gas_price str

Suggested gas price

gas_limit str

Suggested gas limit

from_api_response classmethod

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

Create from API response.

LiFiTransferStatus

Bases: StrEnum

LiFi cross-chain transfer status values.

LiFiTransferSubstatus

Bases: StrEnum

LiFi cross-chain transfer substatus values.

LiFiReceiptParser

LiFiReceiptParser(**kwargs: Any)

Parser for LiFi transaction receipts.

This parser extracts swap/bridge results from transaction receipts by: 1. Checking transaction status 2. Parsing ERC-20 Transfer event logs to find amounts 3. Identifying the wallet's sent and received amounts

For cross-chain transfers, amount_out reflects what was sent to the bridge on the source chain. The actual received amount on the destination chain must be checked via the LiFi status API.

Example

parser = LiFiReceiptParser() receipt = web3.eth.get_transaction_receipt(tx_hash)

result = parser.parse_swap_receipt( receipt=receipt, wallet_address="0x...", token_out="0x...", ) print(f"Sent: {result.amount_in}, Received: {result.amount_out}")

Initialize LiFiReceiptParser.

参数:

名称 类型 描述 默认
**kwargs Any

Keyword arguments passed by the receipt_registry. chain: Chain name for token decimal resolution.

{}

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,
    tool: str | None = None,
    is_cross_chain: bool = False,
) -> LiFiSwapResult

Parse a swap/bridge transaction receipt.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt from web3

必需
wallet_address str

Address that executed the transaction

必需
token_out str

Output token address (on source chain for bridges)

必需
token_in str | None

Input token address (optional)

None
expected_amount_out int | None

Expected output amount for validation

None
tool str | None

Bridge/DEX tool used

None
is_cross_chain bool

Whether this is a cross-chain transfer

False

返回:

类型 描述
LiFiSwapResult

LiFiSwapResult with parsed data

extract_swap_amounts

extract_swap_amounts(
    receipt: dict[str, Any],
) -> SwapAmounts | None

Extract swap amounts for Result Enrichment system.

Called by ResultEnricher after SWAP intent execution.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt

必需

返回:

类型 描述
SwapAmounts | None

SwapAmounts dataclass or None if not found

extract_bridge_data

extract_bridge_data(
    receipt: dict[str, Any],
    *,
    from_chain: str | None = None,
    to_chain: str | None = None,
    token: str | None = None,
    amount: str | Decimal | None = None,
    bridge: str | None = None,
    expected_amount_out: str | Decimal | None = None,
) -> BridgeData | None

Extract typed BridgeData from a LiFi bridge receipt (VIB-3226).

LiFi is primarily a swap aggregator but also handles cross-chain transfers (via embedded bridge "tools" like Across, Hop, etc.). When a BRIDGE intent is routed through LiFi — or a SwapIntent is cross-chain — the Diamond proxy emits ERC-20 Transfers that move the wallet's tokens to the bridge facet. We use the first wallet->contract Transfer to derive amount_sent and the token.

Destination-chain settlement must be tracked via the LiFi status API — destination_tx_hash is intentionally None here.

extract_position_id

extract_position_id(receipt: dict[str, Any]) -> int | None

LiFi swaps do not create LP positions.

extract_liquidity

extract_liquidity(
    receipt: dict[str, Any],
) -> dict[str, Any] | None

LiFi swaps do not provide liquidity events.

extract_lp_close_data

extract_lp_close_data(
    receipt: dict[str, Any],
) -> dict[str, Any] | None

LiFi swaps do not close LP positions.

extract_protocol_fees

extract_protocol_fees(_receipt: dict[str, Any]) -> None

Placeholder for LiFi aggregator protocol-fee extraction (VIB-3204).

LiFi surfaces per-step fees in the quote response at compile time; receipts don't contain a uniform fee event. Threading the quote metadata into parser scope is deferred to a follow-up.

Follow-up ticket: "Protocol fee extraction for aggregators (Enso, LiFi) — follow-up to VIB-3204".

parse_approval_receipt

parse_approval_receipt(
    receipt: dict[str, Any],
) -> dict[str, Any]

Parse an approval transaction receipt.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt from web3

必需

返回:

类型 描述
dict[str, Any]

Dict with approval result