Examples¶
Runnable scripts from the examples/ directory in the repo.
List tokens¶
List all supported tokens, optionally filtered by blockchain.
1"""List all supported tokens, optionally filtered by blockchain."""
2
3import asyncio
4
5from intents import IntentsClient
6
7
8async def main() -> None:
9 async with IntentsClient() as client:
10 tokens = await client.get_tokens()
11
12 print(f"{'Symbol':<12} {'Blockchain':<12} {'Price (USD)':<14} Asset ID")
13 print("-" * 80)
14 for token in sorted(tokens, key=lambda t: (t.blockchain, t.symbol)):
15 price = f"${token.price:.4f}"
16 print(f"{token.symbol:<12} {token.blockchain:<12} {price:<14} {token.asset_id}")
17
18
19if __name__ == "__main__":
20 asyncio.run(main())
Dry quote¶
Preview a swap quote without creating a deposit address — no funds move.
1"""Preview a swap quote without creating a deposit address.
2
3Simulates swapping 0.01 ETH (Ethereum) → USDC (Arbitrum).
4No funds are moved — dry=True means no deposit address is generated.
5"""
6
7import asyncio
8from datetime import datetime, timedelta, timezone
9
10from intents import (
11 DepositType,
12 IntentsClient,
13 QuoteRequest,
14 RecipientType,
15 RefundType,
16 SwapType,
17 find_token,
18 from_atomic,
19 token_to_atomic,
20)
21
22REFUND_ADDRESS = "0x2527D02599Ba641c19FEa793cD0F167589a0f10D"
23RECIPIENT_ADDRESS = "0x2527D02599Ba641c19FEa793cD0F167589a0f10D"
24AMOUNT_ETH = "0.01"
25
26
27async def main() -> None:
28 async with IntentsClient() as client:
29 tokens = await client.get_tokens()
30
31 eth = find_token(tokens, "eth", "ETH")
32 usdc = find_token(tokens, "arb", "USDC")
33
34 response = await client.get_quote(
35 QuoteRequest(
36 dry=True,
37 swap_type=SwapType.EXACT_INPUT,
38 slippage_tolerance=100,
39 origin_asset=eth.asset_id,
40 destination_asset=usdc.asset_id,
41 deposit_type=DepositType.ORIGIN_CHAIN,
42 amount=token_to_atomic(AMOUNT_ETH, eth),
43 refund_to=REFUND_ADDRESS,
44 refund_type=RefundType.ORIGIN_CHAIN,
45 recipient=RECIPIENT_ADDRESS,
46 recipient_type=RecipientType.DESTINATION_CHAIN,
47 deadline=datetime.now(timezone.utc) + timedelta(hours=1),
48 )
49 )
50
51 quote = response.quote
52 print(f"Swap preview: {AMOUNT_ETH} ETH → USDC")
53 print(f" Amount in: {quote.amount_in_formatted} ETH (${quote.amount_in_usd})")
54 print(f" Expected out: {quote.amount_out_formatted} USDC (${quote.amount_out_usd})")
55 print(f" Minimum out: {from_atomic(quote.min_amount_out, usdc.decimals):.6f} USDC")
56 print(f" Time estimate: {quote.time_estimate}s")
57 print(f" Correlation ID: {response.correlation_id}")
58
59
60if __name__ == "__main__":
61 asyncio.run(main())
Swap¶
End-to-end live swap: fetch a real quote, get a deposit address, then poll for execution status.
1"""Execute a real swap: ETH (Ethereum) → USDC (Arbitrum).
2
3Flow:
4 1. Fetch a live quote (dry=False) to get a deposit address.
5 2. Print the deposit instructions.
6 3. Poll the status until the swap completes or fails.
7
8Set INTENTS_JWT in your environment if you have a JWT token.
9"""
10
11import asyncio
12import os
13from datetime import datetime, timedelta, timezone
14
15from intents import (
16 DepositType,
17 IntentsClient,
18 QuoteRequest,
19 RecipientType,
20 RefundType,
21 SwapStatus,
22 SwapType,
23 find_token,
24 token_to_atomic,
25)
26
27REFUND_ADDRESS = "0x2527D02599Ba641c19FEa793cD0F167589a0f10D"
28RECIPIENT_ADDRESS = "0x2527D02599Ba641c19FEa793cD0F167589a0f10D"
29AMOUNT_ETH = "0.01"
30POLL_INTERVAL_SECONDS = 10
31TERMINAL_STATUSES = {SwapStatus.SUCCESS, SwapStatus.REFUNDED, SwapStatus.FAILED}
32
33
34async def main() -> None:
35 async with IntentsClient(jwt_token=os.getenv("INTENTS_JWT")) as client:
36 tokens = await client.get_tokens()
37 eth = find_token(tokens, "eth", "ETH")
38 usdc = find_token(tokens, "arb", "USDC")
39
40 response = await client.get_quote(
41 QuoteRequest(
42 dry=False,
43 swap_type=SwapType.EXACT_INPUT,
44 slippage_tolerance=100,
45 origin_asset=eth.asset_id,
46 destination_asset=usdc.asset_id,
47 deposit_type=DepositType.ORIGIN_CHAIN,
48 amount=token_to_atomic(AMOUNT_ETH, eth),
49 refund_to=REFUND_ADDRESS,
50 refund_type=RefundType.ORIGIN_CHAIN,
51 recipient=RECIPIENT_ADDRESS,
52 recipient_type=RecipientType.DESTINATION_CHAIN,
53 deadline=datetime.now(timezone.utc) + timedelta(hours=1),
54 )
55 )
56
57 quote = response.quote
58 print(f"Send exactly {quote.amount_in_formatted} ETH to:")
59 print(f" Deposit address: {quote.deposit_address}")
60 if quote.deposit_memo:
61 print(f" Memo (required): {quote.deposit_memo}")
62 print(f" Quote expires: {quote.deadline}")
63 print(f" Expected out: {quote.amount_out_formatted} USDC")
64 print()
65
66 print("Polling for swap status...")
67 while True:
68 status_response = await client.get_status(
69 deposit_address=quote.deposit_address,
70 deposit_memo=quote.deposit_memo,
71 )
72 status = status_response.status
73 print(f" [{datetime.now().strftime('%H:%M:%S')}] {status}")
74
75 if status in TERMINAL_STATUSES:
76 break
77
78 await asyncio.sleep(POLL_INTERVAL_SECONDS)
79
80 details = status_response.swap_details
81 if status == SwapStatus.SUCCESS:
82 print(f"\nSwap complete!")
83 print(f" Received: {details.amount_out_formatted} USDC (${details.amount_out_usd})")
84 for tx in details.destination_chain_tx_hashes:
85 print(f" Explorer: {tx.explorer_url}")
86 elif status == SwapStatus.REFUNDED:
87 print(f"\nSwap refunded. Reason: {details.refund_reason}")
88 print(f" Refunded: {details.refunded_amount_formatted} ETH")
89 else:
90 print(f"\nSwap failed.")
91
92
93if __name__ == "__main__":
94 asyncio.run(main())
Submit deposit¶
Notify the 1Click service of an on-chain deposit transaction hash so it processes the swap immediately.
1"""Speed up swap processing by submitting the deposit transaction hash.
2
3After sending funds to the deposit address from a live quote, call this
4to notify the 1Click service immediately rather than waiting for it to
5detect the deposit itself.
6"""
7
8import asyncio
9import os
10
11from intents import IntentsClient, SubmitDepositRequest
12
13
14DEPOSIT_ADDRESS = "0xYourDepositAddressHere"
15TX_HASH = "0xYourTransactionHashHere"
16
17
18async def main() -> None:
19 async with IntentsClient(jwt_token=os.getenv("INTENTS_JWT")) as client:
20 response = await client.submit_deposit(
21 SubmitDepositRequest(
22 tx_hash=TX_HASH,
23 deposit_address=DEPOSIT_ADDRESS,
24 )
25 )
26
27 print(f"Status: {response.status}")
28 print(f"Correlation ID: {response.correlation_id}")
29 print(f"Updated at: {response.updated_at}")
30
31
32if __name__ == "__main__":
33 asyncio.run(main())
Any-input withdrawals¶
Fetch withdrawals for an ANY_INPUT quote, paginating through results.
1"""Fetch withdrawals for an ANY_INPUT quote with pagination."""
2
3import asyncio
4import os
5
6from intents import IntentsClient
7
8DEPOSIT_ADDRESS = "0xYourDepositAddressHere"
9
10
11async def main() -> None:
12 async with IntentsClient(jwt_token=os.getenv("INTENTS_JWT")) as client:
13 response = await client.get_any_input_withdrawals(
14 deposit_address=DEPOSIT_ADDRESS,
15 limit=10,
16 sort_order="desc",
17 )
18
19 print(f"Asset: {response.asset}")
20 print(f"Recipient: {response.recipient}")
21
22 if response.withdrawals is None:
23 print("No withdrawals found.")
24 return
25
26 w = response.withdrawals
27 print(f"\nWithdrawal:")
28 print(f" Status: {w.status}")
29 print(f" Amount: {w.amount_out_formatted} (${w.amount_out_usd})")
30 print(f" Fee: {w.withdraw_fee_formatted} (${w.withdraw_fee_usd})")
31 print(f" Time: {w.timestamp}")
32 print(f" Tx hash: {w.hash}")
33
34
35if __name__ == "__main__":
36 asyncio.run(main())