Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
This page provides a guide to performing efficient token swaps on Solana using Jupiter Exchange, complete with slippage control for precise execution.
quote_url = (
f"{JUP_API}/quote?"
f"inputMint={input_mint}"
f"&outputMint={output_mint}"
f"&amount={int(input_amount * LAMPORTS_PER_SOL)}"
f"&slippageBps={slippage_bps}"
f"&onlyDirectRoutes=true"
f"&maxAccounts=20"
)async with session.post(
f"{JUP_API}/swap",
json={ "quoteResponse": quote_data, "userPublicKey": str(agent.wallet_address), ... },
) as swap_response:
swap_data = await swap_response.json()swap_transaction_buf = base64.b64decode(swap_data["swapTransaction"])
transaction = VersionedTransaction.deserialize(swap_transaction_buf)
transaction.sign([agent.wallet])
signature = await agent.connection.send_raw_transaction(transaction.serialize())await agent.connection.confirm_transaction(
signature, commitment=Confirmed, last_valid_block_height=latest_blockhash.value.last_valid_block_height
)from solders.pubkey import Pubkey
output_mint = Pubkey("So11111111111111111111111111111111111111112") # SOL mint
signature = await TradeManager.trade(agent, output_mint, 10, slippage_bps=100)
print(f"Swap successful: {signature}")Effortlessly deploy new SPL tokens with customized parameters using the TokenDeploymentManager class. This utility simplifies token creation for any project on the Solana blockchain.
lamports = await connection.get_minimum_balance_for_rent_exemption(MINT_LEN) mint = Keypair()
logger.info(f"Generated mint address: {mint.pubkey()}") create_account_ix = create_account(
CreateAccountParams(
from_pubkey=agent.wallet_address,
to_pubkey=mint.pubkey(),
lamports=lamports,
space=MINT_LEN,
program_id=TOKEN_PROGRAM_ID,
)
) initialize_mint_ix = initialize_mint(
program_id=TOKEN_PROGRAM_ID,
mint=mint.pubkey(),
decimals=decimals,
mint_authority=Pubkey.from_string(agent.wallet_address),
freeze_authority=Pubkey.from_string(agent.wallet_address),
) transaction = Transaction()
transaction.add(create_account_ix, initialize_mint_ix)
tx_signature = await agent.connection.send_transaction(
transaction, agent.wallet, mint, opts={"preflight_commitment": Confirmed}
) except Exception as e:
logger.error(f"Token deployment failed: {str(e)}")
raise Exception(f"Token deployment failed: {str(e)}") Efficiently burn SPL tokens and close associated token accounts with the BurnManager class. This functionality is particularly useful for cleaning up unused accounts and managing token supplies.
burn_instruction = burn(
BurnParams(
program_id=TOKEN_PROGRAM_ID,
account=token_account_pubkey,
mint=mint,
owner=owner,
amount=token_balance
)
)
transaction.add(burn_instruction)except Exception as e:
print(f"Error sending transaction for {token_account}: {e}")close_account_instruction = close_account(
CloseAccountParams(
program_id=TOKEN_PROGRAM_ID,
account=token_account_pubkey,
dest=owner,
owner=owner
)
)
transaction.add(close_account_instruction)transaction.add(set_compute_unit_price(100_000))
transaction.add(set_compute_unit_limit(100_000))transaction.sign(agent.wallet)
txn_sig = client.send_transaction(transaction, agent.wallet, opts=TxOpts(skip_preflight=True)).valuefor token_account in token_accounts:
BurnManager.burn_and_close_account(agent, token_account)response = await agent.connection.get_balance(agent.wallet_address, commitment=Confirmed)
return response.value / LAMPORTS_PER_SOL response = await agent.connection.get_token_account_balance(token_address, commitment=Confirmed)
return float(response.value.ui_amount) except Exception as error:
raise Exception(f"Failed to get balance: {str(error)}") from error balance = await BalanceFetcher.get_balance(agent)
print(f"SOL Balance: {balance} SOL") token_address = Pubkey("Your SPL Token Mint Address Here")
balance = await BalanceFetcher.get_balance(agent, token_address)
print(f"SPL Token Balance: {balance}") Effortlessly stake SOL on the Jup validator using the StakeManager class. This guide explains how to execute staking operations with simplicity and precision.
url = f"https://worker.jup.ag/blinks/swap/So11111111111111111111111111111111111111112/jupSoLaHXQiZZTSfEWMTRRgpnyFm8f6sZdosWBjx93v/{amount}"
payload = {"account": str(agent.wallet_address)}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as res:
if res.status != 200:
raise Exception(f"Failed to fetch transaction: {res.status}")
data = await res.json() txn = VersionedTransaction.deserialize(base64.b64decode(data["transaction"]))
latest_blockhash = await agent.connection.get_latest_blockhash()
txn.message.recent_blockhash = latest_blockhash.value.blockhash
txn.sign([agent.wallet]) signature = await agent.connection.send_raw_transaction(
txn.serialize(),
opts={"skip_preflight": False, "max_retries": 3},
)
await agent.connection.confirm_transaction(
signature,
commitment=Confirmed,
last_valid_block_height=latest_blockhash.value.last_valid_block_height,
) except Exception as e:
raise Exception(f"jupSOL staking failed: {str(e)}") Effortlessly obtain SOL from the Solana faucet for development and testing purposes on devnet or testnet environments using the FaucetManager class.
response = await agent.connection.request_airdrop(
agent.wallet_address, 5 * LAMPORTS_PER_SOL
) latest_blockhash = await agent.connection.get_latest_blockhash()
await agent.connection.confirm_transaction(
tx_signature,
commitment=Confirmed,
last_valid_block_height=latest_blockhash.value.last_valid_block_height
) except KeyError:
raise Exception("Airdrop response did not contain a transaction signature.")
except RPCException as e:
raise Exception(f"Faucet request failed: {str(e)}")
except Exception as e:
raise Exception(f"An error occurred: {str(e)}") This page details the steps and functionality to integrate Jupiter's exchange capabilities into your Solana Agents seamlessly.
async def stake_with_jup(agent: SolanaAgentKit, amount: float) -> str:
# Stake SOL using Jupiter's APItry:
# Execute staking logic
except Exception as e:
raise Exception(f"jupSOL staking failed: {str(e)}")transaction_signature = await StakeManager.stake_with_jup(agent, 1.5)
print(f"Transaction successful: {transaction_signature}")try:
transaction_signature = await StakeManager.stake_with_jup(agent, 1.5)
except Exception as error:
print(f"Staking failed: {error}")is_successful = await TokenTransferManager.verify_transfer(transfer_result)async def transfer_native_sol(agent: SolanaAgentKit, to: Pubkey, amount: float) -> str:
# Create and send transaction for native SOL transferasync def transfer_spl_tokens(rpc_client: AsyncClient, agent: SolanaAgentKit, recipient: Pubkey, spl_token: Pubkey, amount: float) -> str:
# Perform token transfer with validation and error handlingasync def confirm_transaction(agent: SolanaAgentKit, signature: str) -> None:
# Confirm transaction on the blockchainasync def execute_transfer(to: Pubkey, amount: float, mint: Optional[Pubkey] = None) -> TransferResult:
# Execute either SOL or SPL token transfer based on parametersawait SolanaTransferHelper.transfer_native_sol(agent, recipient_address, amount)await SolanaTransferHelper.transfer_spl_tokens(rpc_client, agent, recipient_address, token_mint_address, amount)This Python script provides functionality to launch custom SPL tokens on the Solana blockchain through Pump.fun, a platform that enables the creation and management of tokens.
async def _create_token_transaction(
session: aiohttp.ClientSession,
agent: SolanaAgentKit,
mint_keypair: Keypair,
metadata_response: Dict[str, Any],
options: Optional[PumpfunTokenOptions] = None
) -> bytes:async def launch_pumpfun_token(
agent: SolanaAgentKit,
token_name: str,
token_ticker: str,
description: str,
image_url: str,
options: Optional[PumpfunTokenOptions] = None
) -> TokenLaunchResult:async def _upload_metadata(
session: aiohttp.ClientSession,
token_name: str,
token_ticker: str,
description: str,
image_url: str,
options: Optional[PumpfunTokenOptions] = None
) -> Dict[str, Any]:from solana.rpc.async_api import AsyncClient
from Kaze.agent import SolanaAgentKit
# Initialize the agent and other parameters
agent = SolanaAgentKit(client=AsyncClient("https://api.mainnet-beta.solana.com"))
token_name = "MyNewToken"
token_ticker = "MNTK"
description = "This is a test token."
image_url = "https://example.com/token_image.png"
# Optional: Provide additional configuration options
options = PumpfunTokenOptions(
initial_liquidity_sol=1.0, # Amount of SOL for initial liquidity
slippage_bps=100, # Slippage in basis points (1% = 100)
priority_fee=50000 # Priority fee in lamports
)
# Launch the token
launch_result = await PumpfunTokenManager.launch_pumpfun_token(
agent,
token_name,
token_ticker,
description,
image_url,
options
)
# Print the launch result
print(f"Token launched successfully! Mint address: {launch_result.mint}")
print(f"Transaction signature: {launch_result.signature}")except Exception as error:
logger.error(f"Error in launch_pumpfun_token: {error}")
raiseThis guide explains how to execute buy and sell operations on Raydium liquidity pools, utilizing Solana and Kaze AI's toolkit to ensure efficient and secure token trading with slippage control.
pair_address = "Raydium_Pool_Address"
signature = await RaydiumManager.buy_with_raydium(agent, pair_address, sol_in=0.05, slippage=3)
print(f"Transaction Signature: {signature}")pair_address = "Raydium_Pool_Address"
success = await RaydiumManager.sell_with_raydium(agent, pair_address, percentage=50, slippage=2)
print(f"Transaction success: {success}")client = Client(agent.rpc_url)
pool_keys = fetch_pool_keys(pair_address)
amount_in = int(sol_in * SOL_DECIMAL)
base_reserve, quote_reserve, token_decimal = get_token_reserves(pool_keys)
amount_out = sol_for_tokens(sol_in, base_reserve, quote_reserve)
slippage_adjustment = 1 - (slippage / 100)
minimum_amount_out = int(amount_out * slippage_adjustment * 10**token_decimal)token_balance = get_token_balance(str(mint))
token_balance = token_balance * (percentage / 100)
base_reserve, quote_reserve, token_decimal = get_token_reserves(pool_keys)
amount_out = tokens_for_sol(token_balance, base_reserve, quote_reserve)
slippage_adjustment = 1 - (slippage / 100)
amount_out_with_slippage = amount_out * slippage_adjustment
minimum_amount_out = int(amount_out_with_slippage * SOL_DECIMAL)wsol_token_account = Pubkey.create_with_seed(payer_keypair.pubkey(), seed, TOKEN_PROGRAM_ID)
create_wsol_account_instr = create_account_with_seed(...)
init_wsol_account_instr = initialize_account(...)swap_instructions = make_swap_instruction(amount_in, minimum_amount_out, token_account_in, token_account_out, pool_keys, payer_keypair)
compiled_message = MessageV0.try_compile(...)