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 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 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
¶
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
¶
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 supported tokens.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
chain_id
|
int | None
|
Filter by chain ID (optional) |
None
|
返回:
| 类型 | 描述 |
|---|---|
dict[str, Any]
|
Dict of tokens by chain |
get_chains
¶
Get supported chains.
返回:
| 类型 | 描述 |
|---|---|
list[dict[str, Any]]
|
List of chain information dicts |
resolve_chain_id
staticmethod
¶
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 |
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
¶
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
¶
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
¶
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 |
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
¶
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,
)
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 |
from_api_response
classmethod
¶
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 |
from_api_response
classmethod
¶
Create LiFiStep from API response.
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
¶
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
¶
Create from API response.
LiFiTransferStatus
¶
Bases: StrEnum
LiFi cross-chain transfer status values.
LiFiTransferSubstatus
¶
Bases: StrEnum
LiFi cross-chain transfer substatus values.
LiFiReceiptParser
¶
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 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
¶
LiFi swaps do not create LP positions.
extract_liquidity
¶
LiFi swaps do not provide liquidity events.
extract_lp_close_data
¶
LiFi swaps do not close LP positions.
extract_protocol_fees
¶
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 an approval transaction receipt.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
receipt
|
dict[str, Any]
|
Transaction receipt from web3 |
必需 |
返回:
| 类型 | 描述 |
|---|---|
dict[str, Any]
|
Dict with approval result |