import base64
import cobo_waas2
import json
import requests
import uuid
from solders.pubkey import Pubkey
# ========= Jupiter configuration =========
JUPITER_BASE_URL = "https://lite-api.jup.ag/swap/v1/"
SOLANA_RPC = "https://api.mainnet-beta.solana.com"
# Fixed mint addresses for Solana mainnet tokens. Native SOL uses a special system-defined mint address.
SOL_MINT = "So11111111111111111111111111111111111111112"
JLP_MINT = "27G8MtK7VtTcCHkpASjSDdkWWYfoqT6ggEuKidVJidD4"
USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
USDT_MINT = "Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB"
COIN_MINT_MAP = {
'SOL': SOL_MINT,
'SOL_JLP': JLP_MINT,
'SOL_USDC': USDC_MINT,
'SOL_USDT': USDT_MINT,
}
COIN_DECIMALS = {
SOL_MINT: 9,
JLP_MINT: 6,
USDC_MINT: 6,
USDT_MINT: 6,
}
# ========= Jupiter API section (calling Jupiter quote/swap APIs) =========
def get_jupiter_quote(input_mint, output_mint, amount, slippage_bps=50):
"""
Request a swap quote from Jupiter.
Args:
input_mint: Mint address of the input token
output_mint: Mint address of the output token
amount: Swap amount
slippage_bps: Slippage tolerance
"""
params = {
"inputMint": input_mint,
"outputMint": output_mint,
"amount": str(amount),
"slippageBps": str(slippage_bps)
}
r = requests.get(JUPITER_BASE_URL + "quote", params=params, verify=False)
r.raise_for_status()
return r.json()
def get_swap_instructions(quote, address):
"""
Generate swap instructions based on a Jupiter quote.
"""
payload = {
"quoteResponse": quote,
"userPublicKey": address,
"dynamicSlippage": True,
"useSharedAccounts": True,
"dynamicComputeUnitLimit": True,
"wrapAndUnwrapSol": True,
"skipUserAccountsRpcCalls": False,
}
r = requests.post(JUPITER_BASE_URL + "swap-instructions", json=payload, verify=False)
r.raise_for_status()
return r.json()
def fetch_address_lookup_table(alt_address):
"""
Fetch accounts inside the Address Lookup Table (ALT).
Args:
alt_address: ALT address
Returns:
list: List of addresses inside the ALT
"""
print(f" Fetching ALT: {alt_address}")
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "getAccountInfo",
"params": [
alt_address,
{"encoding": "base64"}
]
}
try:
response = requests.post(SOLANA_RPC, json=payload, verify=False)
response.raise_for_status()
result = response.json()
if "result" not in result or not result["result"] or not result["result"]["value"]:
print(f" Warning: Unable to fetch ALT account info: {alt_address}")
return []
# Parsing ALT account information
account_data = result["result"]["value"]["data"][0]
raw_data = base64.b64decode(account_data)
# ALT account layout:
# - The first 56 bytes are header metadata
# - Every subsequent 32-byte chunk is a public key inside the lookup table
addresses = []
offset = 56 # Skip the header
while offset + 32 <= len(raw_data):
pubkey_bytes = raw_data[offset:offset + 32]
pubkey = Pubkey(pubkey_bytes)
addresses.append(str(pubkey))
offset += 32
print(f" Found {len(addresses)} accounts in ALT")
return addresses
except Exception as e:
print(f" ALT fetch failed: {e}")
return []
# ========= Cobo API section =========
class CoboAPI:
def __init__(self, api_private_key, host):
self.api_private_key = api_private_key
self.host = host
def get_configuration(self):
return cobo_waas2.Configuration(
api_private_key=self.api_private_key,
host=self.host
)
def swap_coin(self, wallet_id, address, input_coin, output_coin, amount):
"""
Call Cobo WaaS service to execute a Solana contract call for a token swap.
Args:
wallet_id: Cobo wallet ID
address: Solana address
input_coin: Input token symbol
output_coin: Output token symbol
amount: Swap amount
Returns:
transaction_id: Transaction ID in Cobo Portal
"""
print(f"Executing token swap via Cobo WaaS... address={address}, {input_coin} -> {output_coin}, amount={amount}")
input_mint = COIN_MINT_MAP[input_coin]
output_mint = COIN_MINT_MAP[output_coin]
amount = self.to_jupiter_amount(amount, COIN_DECIMALS[input_mint])
with cobo_waas2.ApiClient(self.get_configuration()) as api_client:
# 1. Get a quote from Jupiter
quote = get_jupiter_quote(input_mint, output_mint, amount)
# 2. Get swap instructions from Jupiter
instructions_json = get_swap_instructions(quote, address)
# 3. Build transaction instructions and ALT accounts
instructions, alt_accounts = self.build_transaction(instructions_json, address)
# 4. Call Cobo WaaS service to create a contract call transaction
contract_call_params = cobo_waas2.ContractCallParams(
request_id=str(uuid.uuid4()),
chain_id="SOL",
source=cobo_waas2.ContractCallSource(
cobo_waas2.MpcContractCallSource(
source_type=cobo_waas2.ContractCallSourceType.ORG_CONTROLLED,
wallet_id=wallet_id,
address=address,
)
),
destination=cobo_waas2.ContractCallDestination(
cobo_waas2.SolContractCallDestination(
destination_type=cobo_waas2.ContractCallDestinationType.SOL_CONTRACT,
instructions=instructions,
address_lookup_table_accounts=alt_accounts
)
),
)
print("=================================")
print("contract call params:", contract_call_params.to_json())
print("=================================")
api_instance = cobo_waas2.TransactionsApi(api_client)
try:
api_response = api_instance.create_contract_call_transaction(
contract_call_params=contract_call_params
)
print("contract call response:", api_response)
return api_response.transaction_id
except Exception as e:
print("Error:", e)
return ""
@staticmethod
def build_instruction(ins, address):
accounts = []
print("build instruction:", ins.get("programId"), "accounts:", len(ins.get("accounts", [])))
for acc in ins["accounts"]:
accounts.append(
cobo_waas2.SolContractCallAccount(
pubkey=acc["pubkey"],
is_signer=acc["isSigner"],
is_writable=acc["isWritable"] or acc["pubkey"] == address,
)
)
return cobo_waas2.SolContractCallInstruction(
program_id=ins["programId"],
accounts=accounts,
data=ins["data"],
)
def build_transaction(self, instructions_json, address):
print("build transaction:", json.dumps(instructions_json))
all_instructions = []
for key in ["computeBudgetInstructions", "setupInstructions"]:
all_instructions += [self.build_instruction(i, address) for i in instructions_json.get(key, [])]
if "swapInstruction" in instructions_json and instructions_json["swapInstruction"]:
all_instructions.append(self.build_instruction(instructions_json["swapInstruction"], address))
if "cleanupInstruction" in instructions_json and instructions_json["cleanupInstruction"]:
all_instructions.append(self.build_instruction(instructions_json["cleanupInstruction"], address))
if "otherInstructions" in instructions_json and instructions_json["otherInstructions"]:
all_instructions += [self.build_instruction(i, address) for i in instructions_json["otherInstructions"]]
alt_accounts = []
if "addressLookupTableAddresses" in instructions_json:
alt_addresses = instructions_json["addressLookupTableAddresses"]
for alt_address in alt_addresses:
addresses = fetch_address_lookup_table(alt_address)
if addresses:
alt_accounts.append(
cobo_waas2.SolContractCallAddressLookupTableAccount(
alt_account_key=alt_address,
addresses=addresses,
)
)
return all_instructions, alt_accounts
@staticmethod
def to_jupiter_amount(readable_amount, decimals):
"""Convert human-readable amount to Jupiter integer amount."""
return int(readable_amount * (10 ** decimals))
@staticmethod
def to_readable_amount(raw_amount, decimals):
"""Convert Jupiter integer amount to human-readable format."""
return raw_amount / (10 ** decimals)
if __name__ == "__main__":
# ======== Parameters you must modify for your own environment (required) ========
# TODO: Use the development environment. For production environment, change to "https://api.cobo.com/v2".
cobo_host = "https://api.dev.cobo.com/v2"
# TODO: Enter your API key
api_key = ""
# TODO: Enter your Wallet ID
mpc_wallet_id = ""
# TODO: Enter your Solana address, and make sure it has enough SOL and the token to be swapped
sol_address = ""
# ======== Example: SOL -> USDT ========
myCoboApi = CoboAPI(api_key, cobo_host)
# "SOL" is the token ID of input token, "SOL_USDT" is the token ID of output token (USDT), 0.0001 is the swap amount
myCoboApi.swap_coin(mpc_wallet_id, sol_address, "SOL", "SOL_USDT", 0.0001)
# If you want to swap to JLP or USDC, use the following examples:
# myCoboApi.swap_coin(mpc_wallet_id, sol_address, "SOL", "SOL_JLP", 0.0001)
# myCoboApi.swap_coin(mpc_wallet_id, sol_address, "SOL", "SOL_USDC", 0.0001)
